百度360必应搜狗淘宝本站头条
wireshark怎么抓包c++ base64for_eachwxsqlite3qt sleep
当前位置：网站首页 > 技术分析 > 正文

python多进程编程

liebian365 2025-02-07 18:13 5 浏览 0 评论

fork

windows中是没有fork函数的，一开始直接在Windows中测试，直接报错

import os
import time

ret = os.fork()
if ret == 0:
    while True:
        print("-----1----")
        time.sleep(1)
else:
    while True:
        print("-----2-----")
        time.sleep(1)

AttributeError: module 'os' has no attribute 'fork'

然后在Linux上实验，可以。父子进程执行顺序是不定的。

fork函数调动一次，有两个返回值，父进程返回值>0，子进程返回值=0；

import os
ret = os.fork()
print(ret)

父进程中fork的返回值就是创建的子进程的pid。

import os
ret = os.fork()
print(ret)
if ret>0:
    print("--父进程pid--%d"%os.getpid())
else:
    print("--子进程pid--%d--父进程---%d"%(os.getpid(),os.getppid()))

父子进程的退出

父进程可以在子进行前退出。

import os
import time

ret = os.fork()

if ret==1:
    print("子进程1")
    time.sleep(5)
    print("子进程2")
else:
    print("父进程")
    time.sleep(3)

print("over")

全局变量问题：

变量在父子进程间不会共享，子进程会自己创建一份。

import os
import time

num = 100
ret = os.fork()
if ret==0:
    print("----process1----")
    num += 1
    print("-----process1---num=%d"%(num))

else:
    time.sleep(3)
    print("---process2----num=%d"%(num))

多次fork问题:

import os
import time

ret = os.fork()
if ret==0:
    print("--1--")
else:
    print("--2--")
ret = os.fork()
if ret ==0:
    print("--11--")
else:
    print("--22--")

3个fork问题：

Process创建子进程

multiprocessing模块中Process类

rom multiprocessing import Process
import time

def test():
    while True:
        print("--test--")
        time.sleep(1)

p = Process(target=test)//创建一个进程对象，test函数结束了，这个进程就结束了
p.start()

while True:
    print("--main()--")
    time.sleep(1)

主进程等待子进程先结束

from multiprocessing import Process
import time

def test():
    while True:
        print("--test--")
        time.sleep(1)

p = Process(target=test)
p.start()
#这里主进程不会直接结束，它会等待子进程结束了再结束

Process的init函数

def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):

给target函数传递参数

args：位置参数元祖

kwargs：关键字参数字典

from multiprocessing import Process
import os

def test(name1,name2):
    print("name1=%s,name2=%s"%(name1,name2))

p=Process(target=test,args=("hy","ym"))

p.start()

join阻塞

join([timeout])：等待子进程结束，或等待多少秒（等待时间到了，就不等了）

terminate()：不管任务是否完成，立即终止进程

from multiprocessing import Process
import time

def test():
    for i in range(5):
        print("-%d--"%(i))
        time.sleep(1)

p = Process(target=test)
p.start()

p.join()#阻塞，等待子进程结束

print("--main--")

Process子类创建进程

from multiprocessing import Process
import time

class MyProcess(Process):
    def run(self):
        while True:
            print("--1--")
            time.sleep(1)

p = MyProcess()
p.start()#当没有给process传递target时，会自动执行run函数

while True:
    print("--main--")
    time.sleep(1)

对一个不包含target属性的Process类执行start方法，会运行这个类中的run方法。

进程池Pool

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态生成多个进程。但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以利用multiprocessing模块提供的Pool方法。

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程来执行该请求；如果池中的进程数已经达到指定的最大值，那么请求就会等待，直到池中有进程结束，才会创建新的进程来执行。

from multiprocessing import Pool
import time
import os

def Worker(msg):
    print("%d,%d进程正在执行"%(msg,os.getpid()))
    time.sleep(3)


po=Pool(3)#3表示，进程池最多有3个进程一起执行
for i in range(10):
    #向进程添加任务，如果任务超过进程池中进程个数，需要等待，有任务执行完成，会继续执行。
    po.apply_async(Worker,(i,))


po.close()#关闭进程池，相当于不能添加新任务了
po.join()#主进程默认不会等待进程池中任务执行完成才推出，所以这里要阻塞，不然主进程结束了，进程池中任务没有执行完成也会跟着结束。

上面的代码在Windows中运行时，需要加一个

if __name__ == "__main__":

不然会报错。

pool中apply阻塞方式

from multiprocessing import Pool
import time
import os

def Worker(msg):
    print("%d,%d进程正在执行"%(msg,os.getpid()))
    time.sleep(3)


po=Pool(3)
for i in range(10):
    print(i)
    po.apply(Worker,(i,))#主进程在这里阻塞，等待一个任务的完成


po.close()
po.join()

进程间通信 :

Queue

In [1]: from multiprocessing import Queue

In [2]: q = Queue(3)//队列最大长度为3，只能放三个数据，存放类型随意

In [3]: q.empty()
Out[3]: True

In [4]: q.full()
Out[4]: False

In [5]: q.put("hy")

In [6]: q.put(123)

In [7]: q.get()
Out[7]: 'hy'

In [8]: q.get()
Out[8]: 123

In [9]: q.qsize()
Out[9]: 0

from multiprocessing import Queue,Process
import time

def write(q):
    for value in ["A","B","c"]:
        print("将数据%s放入队列"%value)
        q.put(value)
        time.sleep(1)


def read(q):
    while True:
        if not q.empty():
            value=q.get();
            print("将数据%s从队列中取出"%value)
            time.sleep(1)
        else:
            break;

if __name__ == "__main__":
    q=Queue()#这里没有指明队列长度，表示长度无限
    pw = Process(target=write,args=(q,))
    pr = Process(target=read,args=(q,))
    pw.start()
    pw.join()
    pr.start()
    pr.join()

    print("所有数据以及写读完成")

当使用pool时需要用Manager中的Queue

from multiprocessing import Manager,Pool
import os

def write(q):
    print("wirte启动(%d),其父进程为(%d)"%(os.getpid(),os.getppid()))
    for i in "huangtieniu":
        q.put(i)


def read(q):
    print("read进程启动(%d),其父进程为(%d)"%(os.getpid(),os.getppid()))
    for i in range(q.qsize()):
        print("read从进程中获得消息:%s"%q.get())


if __name__ == "__main__":
    q = Manager().Queue()
    po = Pool()
    po.apply(write,(q,))
    po.apply(read,(q,))
    po.close()
    po.join()
    print("End")

线程：

Python中thread模块是比较底层的模块，Python中的threading模块是对thread做了一层封装的，可以更加方便的使用。

主线程会等待子线程执行完毕再结束。是为了回收子线程的资源

from threading import Thread
import time

#多个线程执行的是同一个函数，相互之间不影响
def test():
    print("I'm so sorry.")
    time.sleep(1)


for i in range(5):
    t = Thread(target=test)
    t.start()

使用Thread子类创建多线程

import threading
import time

class MyThread(threading.Thread):
    def run(self):
        for i in range(3):
            print("I'm %s @ %d"%(self.name,i))
            time.sleep(1)


if __name__ == "__main__":
    t = MyThread()
    t.start()

线程的执行顺序也是不固定的，由操作系统说了算。

import threading
import time

def test1():
    for i in range(3):
        print("----%d------"%i)
        time.sleep(1)

def test2():
    for i in range(4,7):
        print("-----%d-----"%i)
        time.sleep(1)

if __name__ == "__main__":
    t1 = threading.Thread(target=test1)
    t2 = threading.Thread(target=test2)
    t1.start()
    t2.start()

线程共享全局变量，可以利用全局变量通信

from threading import Thread
import time

def work1():
    global num
    num +=1
    print("work1中num=%d"%(num))

def work2():
    global num
    print("work2中num=%d"%(num))

#线程之间会共享全局变量
num = 100

if __name__ == "__main__":
    t1 = Thread(target=work1)
    t2 = Thread(target=work2)
    t1.start()
    time.sleep(1)
    t2.start()

线程共享全局变量可能的问题：

from threading import Thread
import time

def test1():
    global num
    for i in range(1000000):
        num += 1
    print("--test1--num=%d"%(num))

def test2():
    global num
    for i in range(1000000):
        num += 1
    print("--test2--num=%d"%(num))


num = 0

if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test2)
    p1.start()
    time.sleep(3)
    p2.start()

    print("--主线程中num=%d"%(num))

time.sleep(3)这一行注释掉时结果是前一个，加上时结果是后一个。

列表当做实参传递给线程

import threading import Thread
import time

def work1(num):
    num.append(44)
    print("--in work1--",num)

def work2(num):
    time.sleep(1)#保证work1执行完成
    print("--in work2--",num)

num = [11,22,33]

if __name__ == "__main__":
    t1 = Thread(target=work1,args=(num,))
    t2 = Thread(target=work2,args=(num,))
    t1.start()
    t2.start()

避免全局变量被修改的方式

1.加一个flag标识，轮训

from threading import Thread
import time



def test1():
    global num
    global flag
    if flag == 0:
        for i in range(1000000):
            num += 1
        flag = 1
    print("--test1--num=%d"%(num))

def test2():
    global num
    #轮训，太耗费cpu
    while True:
        if flag != 0:
            for i in range(1000000):
                num += 1
            break
    print("--test2--num=%d"%(num))


num = 0
flag = 0

if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test2)
    p1.start()
    #time.sleep(3)
    p2.start()

    print("--主线程中num=%d"%(num))

避免全局变量被修改---互斥锁

某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；知道该线程释放资源，将资源的状态变为“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

from threading import Thread,Lock
import time



def test1():
    global num
    #加锁，test1和test2竞争加锁，如果有一方加锁，另一方会阻塞，知道这个所被解开
    mutex.acquire()
    for i in range(1000000):
        num += 1
    #解锁
    mutex.release()
    print("--test1--num=%d"%(num))

def test2():
    global num
    mutex.acquire()
    for i in range(1000000):
        num += 1
    mutex.release()
    print("--test2--num=%d"%(num))


num = 0
flag = 0

#创建一把互斥锁，默认是没有上锁的
mutex = Lock()

if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test2)
    p1.start()
    #time.sleep(3)
    p2.start()

    print("--主线程中num=%d"%(num))

加锁原则：

能不加锁就不加锁，加锁的范围能满足要求就好，不要太大。等待解锁的方式：通知，不是轮训。

多个线程使用非全局变量，非全局变量各自是各自的，不需要加锁

from threading import Thread
import time
import threading


def test1():
    name = threading.current_thread().name
    print("thread name is %s"%(name))
    num = 100
    if name == "Thread-1":
        num += 1
    else:
        time.sleep(2)
    print("thread is %s,num is %d"%(name,num))



if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test1)
    p1.start()
    p2.start()

死锁

在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。

import time
import threading
from threading import Lock

class MyThread1(threading.Thread):
    def run(self):
        if mutexA.acquire():
            print(self.name+"已对A加锁")
            time.sleep(1)

            if mutexB.acquire():
                print(self.name+"已对B加锁")
            mutexB.release()
        mutexA.release()

class MyThread2(threading.Thread):
    def run(self):
        if mutexB.acquire():
            print(self.name+"已对B加锁")
            time.sleep(1)

            if mutexA.acquire():
                print(self.name+"已对A加锁")
            mutexA.release()
        mutexB.release()

mutexA = threading.Lock()
mutexB = threading.Lock()

if __name__ == "__main__":
    t1 = MyThread1()
    t2 = MyThread2()
    t1.start()
    t2.start()

相关推荐

4万多吨豪华游轮遇险 竟是因为这个原因……

（观察者网讯）4.7万吨豪华游轮搁浅，竟是因为油量太低？据观察者网此前报道，挪威游轮“维京天空”号上周六（23日）在挪威近海发生引擎故障搁浅。船上载有1300多人，其中28人受伤住院。经过数天的调...

“菜鸟黑客”必用兵器之“渗透测试篇二”

"菜鸟黑客"必用兵器之"渗透测试篇二"上篇文章主要针对伙伴们对"渗透测试"应该如何学习？"渗透测试"的基本流程？本篇文章继续上次的分享，接着介绍一下黑客们常用的渗透测试工具有哪些？以及用实验环境让大家...

科幻春晚丨《震动羽翼说“Hello”》两万年星间飞行，探测器对地球的最终告白

作者|藤井太洋译者|祝力新【编者按】2021年科幻春晚的最后一篇小说，来自大家喜爱的日本科幻作家藤井太洋。小说将视角放在一颗太空探测器上，延续了他一贯的浪漫风格。...

麦子陪你做作业(二):KEGG通路数据库的正确打开姿势

作者：麦子KEGG是通路数据库中最庞大的，涵盖基因组网络信息，主要注释基因的功能和调控关系。当我们选到了合适的候选分子，单变量研究也已做完，接着研究机制的时便可使用到它。你需要了解你的分子目前已有哪些...

知存科技王绍迪:突破存储墙瓶颈，详解存算一体架构优势

智东西（公众号：zhidxcom）编辑|韦世玮智东西6月5日消息，近日，在落幕不久的GTIC2021嵌入式AI创新峰会上，知存科技CEO王绍迪博士以《存算一体AI芯片：AIoT设备的算力新选择》...

每日新闻播报(September 14)_每日新闻播报英文

AnOscarstatuestandscoveredwithplasticduringpreparationsleadinguptothe87thAcademyAward...

香港新巴城巴开放实时到站数据 供科技界研发使用

中新网3月22日电据香港《明报》报道，香港特区政府致力推动智慧城市，鼓励公私营机构开放数据，以便科技界研发使用。香港运输署21日与新巴及城巴（两巴）公司签署谅解备忘录，两巴将于2019年第3季度，开...

5款不容错过的APP: Red Bull Alert，Flipagram，WifiMapper

本周有不少非常出色的app推出，鸵鸟电台做了一个小合集。亮相本周榜单的有WifiMapper's安卓版的app，其中包含了RedBull的一款新型闹钟，还有一款可爱的怪物主题益智游戏。一起来看看我...

Qt动画效果展示_qt显示图片

今天在这篇博文中，主要实践Qt动画，做一个实例来讲解Qt动画使用，其界面如下图所示(由于没有录制为gif动画图片，所以请各位下载查看效果)：该程序使用应用程序单窗口，主窗口继承于QMainWindow...

如何从0到1设计实现一门自己的脚本语言

作者：dong...

三年级语文上册 仿写句子 需要的直接下载打印吧

描写秋天的好句好段1.秋天来了，山野变成了美丽的图画。苹果露出红红的脸庞，梨树挂起金黄的灯笼，高粱举起了燃烧的火把。大雁在天空一会儿写“人”字，一会儿写“一”字。2.花园里，菊花争奇斗艳，红的似火，粉...

C++|那些一看就很简洁、优雅、经典的小代码段

目录0等概率随机洗牌：1大小写转换2字符串复制...

二年级上册语文必考句子仿写，家长打印，孩子照着练

二年级上册语文必考句子仿写，家长打印，孩子照着练。具体如下：...

一年级语文上 句子专项练习(可打印)

...

亲自上阵!C++ 大佬深度“剧透”:C++26 将如何在代码生成上对抗 Rust?

...

取消回复欢迎 你 发表评论:

一周热门
最近发表
标签列表
Copyright Your WebSite.Some Rights Reserved.