Python3 多线程多进程

Posted 2020-10-30 msc1

python中的线程是假线程，不同线程之间的切换是需要耗费资源的，因为需要存储线程的上下文，不断的切换就会耗费资源。。

python多线程适合io操作密集型的任务（如socket server 网络并发这一类的）；
python多线程不适合cpu密集操作型的任务，主要使用cpu来计算，如大量的数学计算。
那么如果有cpu密集型的任务怎么办，可以通过多进程来操作（不是多线程）。
假如CPU有8核，每核CPU都可以用1个进程，每个进程可以用1个线程来进行计算。
进程之间不需要使用gil锁，因为进程是独立的，不会共享数据。
进程可以起很多个，但是8核CPU同时只能对8个任务进行操作。

多进程

#测试多进程

import multiprocessing
import time

def run(name):
    time.sleep(2)
    print (‘heelo‘,name)

if __name__ == ‘__main__‘:

    for i in range(10): #起了10个进程
        p = multiprocessing.Process(target=run,args=(‘msc%s‘ %i,))
        p.start()

执行结果：
hello msc1
hello msc0
hello msc2
hello msc3
hello msc5
hello msc4
hello msc6
hello msc7
hello msc8
hello msc9

import multiprocessing
import time,threading

def thread_run():
    print (threading.get_ident()) #get_ident获取当前线程id

def run(name):
    time.sleep(2)
    print (‘hello‘,name)
    t = threading.Thread(target=thread_run,)    #在每个进程中又起了1个线程
    t.start()

if __name__ == ‘__main__‘:

    for i in range(10):     #起了10个进程
        p = multiprocessing.Process(target=run,args=(‘msc%s‘ %i,))
        p.start()

执行结果：
hello msc0
13996
hello msc2
14208
hello msc1
13964
hello msc3
14012
hello msc6
15192
hello msc7
15136
hello msc8
7036
hello msc4
12344
hello msc9
15332
hello msc5
13616

from multiprocessing import Process
import os

def info(title):
    print(title)
    print(‘module name:‘, __name__)
    print(‘parent process:‘, os.getppid())  #获取父进程的id
    print(‘process id:‘, os.getpid())   #获取自身的id
    print("\n\n")

def f(name):
    info(‘\033[31;1mfunction f\033[0m‘)
    print(‘hello‘, name)

if __name__ == ‘__main__‘:
    info(‘\033[32;1mmain process line\033[0m‘)  ##直接调用函数
    # p = Process(target=f, args=(‘bob‘,))
    # p.start()
    # p.join()

执行结果：
main process line
module name: __main__
parent process: 7172   #父进程就是python
process id: 14880       #这个子进程就是python的代码程序

##每个进程都会有一个父进程。

from multiprocessing import Process
import os

def info(title):
    print(title)
    print(‘module name:‘, __name__)
    print(‘parent process:‘, os.getppid())  #获取父进程的id
    print(‘process id:‘, os.getpid())   #获取自身的id
    print("\n\n")

def f(name):
    info(‘\033[31;1mcalled from child process function f\033[0m‘)
    print(‘hello‘, name)

if __name__ == ‘__main__‘:
    info(‘\033[32;1mmain process line\033[0m‘)
    p = Process(target=f, args=(‘msc‘,))    #设置子进程
    p.start()   #启动子进程
    # p.join()

执行结果：
main process line
module name: __main__
parent process: 1136    #主进程pycharm
process id: 14684       #子进程python代码

called from child process function f
module name: __mp_main__
parent process: 14684   #主进程python代码（1136的子进程）
process id: 15884       #python代码（主进程14684）中的子进程的子15884
## 每个进程都有主进程（父进程）

hello msc

进程间通讯

默认进程之间数据是不共享的，如果一定要实现互访可以通过Queue来实现，这个Queue和线程中的Queue使用方法一样，不过线程中的Queue只能在线程之间使用。

线程

import queue
import threading

def f():
    q.put([42,None,‘heelo‘])

if __name__ == ‘__main__‘:
    q = queue.Queue()     
    p = threading.Thread(target=f,)

    p.start()

    print (q.get())
    p.join()

执行结果：
[42, None, ‘heelo‘]
## 通过子线程put进去数据，然后在主线程get出内容，表明线程之间数据是可以共享的。

import queue
from multiprocessing import Process

def f():
    q.put([66,None,‘hello‘])    #这里的q属于主进程

if __name__ == ‘__main__‘:
    q = queue.Queue()   #主进程起的q
    p = Process(target=f,)
    ## 在主进程中来定义子进程；如果在主进程中启动了子进程，那么主进程和子进程之间内存是独立的。
    ## 因为内存独立，子进程p是无法访问主进程def f()中的q的。
    p.start()

    print (q.get())
    p.join()

执行结果：
Process Process-1:
Traceback (most recent call last):
  File "C:\Python35\lib\multiprocessing\process.py", line 249, in _bootstrap
    self.run()
  File "C:\Python35\lib\multiprocessing\process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "D:\Program Files (x86)\python\day31\test.py", line 7, in f
    q.put([66,None,‘hello‘])    #这里的q属于主进程
NameError: name ‘q‘ is not defined

##可以看到已经报错，这是因为子进程不能访问主进程的q

import queue
from multiprocessing import Process

def f(qq):
    qq.put([66,None,‘hello‘])

if __name__ == ‘__main__‘:
    q = queue.Queue()
    p = Process(target=f,args=(q,)) #将父进程q传给子进程

    p.start()

    print (q.get())
    p.join()

执行结果：
Traceback (most recent call last):
  File "D:/Program Files (x86)/python/day31/test.py", line 13, in <module>
    p.start()
  File "C:\Python35\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "C:\Python35\lib\multiprocessing\context.py", line 212, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\Python35\lib\multiprocessing\context.py", line 313, in _Popen
    return Popen(process_obj)
  File "C:\Python35\lib\multiprocessing\popen_spawn_win32.py", line 66, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\Python35\lib\multiprocessing\reduction.py", line 59, in dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: can‘t pickle _thread.lock objects

## 这是因为我们将线程的q传给另一个进程，这是不可以的，线程只属于当前进程，不能传给其他进程。
## 如果想将q传给子进程，那么必须将进程q传进去，而不是线程q。

from multiprocessing import Process,Queue
##大写的Queue是进程队列； queue是线程队列
##大写的Queue需要从multiprocessing导入

def f(qq):
    qq.put([66,None,‘hello‘])

if __name__ == ‘__main__‘:
    q = Queue()
    p = Process(target=f,args=(q,)) #将父进程q传给子进程

    p.start()

    print (q.get()) #父进程去get子进程的内容
    p.join()

执行结果：

[66, None, ‘hello‘]

##父进程可以get子进程put进去的内容了；从表面上看感觉是两个进程共享了数据，其实不然。

‘‘‘ 
 现在已经实现了进程间的通讯。父进程将q传给子进程，其实是克隆了一份q给子进程，此时子进程就多了一个q进程队列；
 但是父进程又为什么能够get子进程put进去的数据呢，这是因为当前两个进程在内存空间依然是独立的，只不过子进程put的数据 通过pickle序列化放到内存中一个中间的位置，然后父进程从这个中间的位置取到数据（而不是从子进程中取的数据）。 
 所以进程间的通讯不是共享数据，而是一个数据的传递。
‘‘‘

进程之间的数据还可以通过管道的方式来通讯

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([66, None, ‘hello from child1‘])  #发送数据给parent_conn
    conn.close()    #发完数据需要关闭

if __name__ == ‘__main__‘:
    parent_conn, child_conn = Pipe()
    ## 生成管道。 生成时会产生两个返回对象，这两个对象相当于两端的电话，通过管道线路连接。
    ## 两个对象分别交给两个变量。
    p = Process(target=f, args=(child_conn,))   #child_conn需要传给对端，用于send数据给parent_conn
    p.start()
    print(parent_conn.recv())  #parent_conn在这端，用于recv数据
    p.join()

执行结果：
[66, None, ‘hello from child1‘]

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([66, None, ‘hello from child1‘])
    conn.send([66, None, ‘hello from child2‘])  #发送两次数据
    conn.close()

if __name__ == ‘__main__‘:
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())  
    p.join()

执行结果：
[66, None, ‘hello from child1‘]
## 可以看到这端只接收到了一次数据

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([66, None, ‘hello from child1‘])
    conn.send([66, None, ‘hello from child2‘])  
    conn.close()

if __name__ == ‘__main__‘:
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    print(parent_conn.recv())   #第二次接收数据
    p.join()

执行结果：
[66, None, ‘hello from child1‘]
[66, None, ‘hello from child2‘]
##对端发送几次，这端就需要接收几次

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([66, None, ‘hello from child1‘])
    conn.send([66, None, ‘hello from child2‘])  #发送两次数据
    conn.close()

if __name__ == ‘__main__‘:
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    print(parent_conn.recv())
    print(parent_conn.recv())   #对端发送两次，本段接收三次
    p.join()

执行结果：
[66, None, ‘hello from child1‘]
[66, None, ‘hello from child2‘]
## 程序卡主了，除非对端在发送一次数据。

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([66, None, ‘hello from child1‘])
    conn.send([66, None, ‘hello from child2‘])  #发送两次数据
    print (conn.recv()) #接收数据
    conn.close()

if __name__ == ‘__main__‘:
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    print(parent_conn.recv())
    parent_conn.send("data from parent_conn")   #发送数据
    p.join()

执行结果：
[66, None, ‘hello from child1‘]
[66, None, ‘hello from child2‘]
data from parent_conn

##通过管道实现了相互发送接收数据(实现了数据传递）

进程间数据交互及共享

from multiprocessing import Process, Manager
import os

def f(d, l, n):
    d[1] = ‘1‘  #放入key和value到空字典中
    d[‘2‘] = 2
    d[0.25] = None

    l.append(n) #将每个进程的n值放入列表中；每个进程的n值都不同。
    print(l)

if __name__ == ‘__main__‘:
    with Manager() as manager:  #做一个别名，此时manager就相当于Manager()
        d = manager.dict()  #生成一个可在多个进程之间传递和共享的字典

        l = manager.list(range(5))  #生成一个可在多个进程之间传递和共享的列表;通过range(5)给列表中生成5个数据
        p_list = []
        for i in range(10): #生成10个进程
            p = Process(target=f, args=(d, l, i))  #将字典和列表传给每个进程，每个进程可以进行修改
            p.start()
            p_list.append(p)    # 将每个进程放入空列表中
        for res in p_list:
            res.join()

        print(d)    #所有进程都执行完毕后打印字典
        print(l)    #所有进程都执行完毕后打印列表

执行结果：
#列表生成的时候自动加入了0-4这5个数；然后每个进程又把各自的n值加入列表
[0, 1, 2, 3, 4, 2]
[0, 1, 2, 3, 4, 2, 3]
[0, 1, 2, 3, 4, 2, 3, 4]
[0, 1, 2, 3, 4, 2, 3, 4, 1]
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0]
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0, 5]
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0, 5, 6]
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0, 5, 6, 7]
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0, 5, 6, 7, 8]
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0, 5, 6, 7, 8, 9] #第十个进程把每个进程添加的n值都加入到列表
{0.25: None, 1: ‘1‘, ‘2‘: 2}  #最后打印的字典
[0, 1, 2, 3, 4, 2, 3, 4, 1, 0, 5, 6, 7, 8, 9] #最后打印的列表

Process finished with exit code 0

from multiprocessing import Process, Manager
import os

def f(d, l):
    d[os.getpid()] = os.getpid()

    l.append(os.getpid())
    print(l)

if __name__ == ‘__main__‘:
    with Manager() as manager:
        d = manager.dict()  #对字典做个调整，也将pid加入到字典中

        l = manager.list(range(5))
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()

        print(d)
        print(l)

执行结果：
[0, 1, 2, 3, 4, 2240]
[0, 1, 2, 3, 4, 2240, 10152]
[0, 1, 2, 3, 4, 2240, 10152, 10408]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156, 6184]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156, 6184, 16168]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156, 6184, 16168, 11384]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156, 6184, 16168, 11384, 15976]
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156, 6184, 16168, 11384, 15976, 16532]
{2240: 2240, 10152: 10152, 10408: 10408, 6312: 6312, 17156: 17156, 6184: 6184, 16168: 16168, 11384: 11384, 15976: 15976, 16532: 16532}
[0, 1, 2, 3, 4, 2240, 10152, 10408, 6312, 17156, 6184, 16168, 11384, 15976, 16532]

##现在我们看到可以实现进程间的数据共享、修改和传递。
##Manager()自带锁，会控制进程之间同一时间修改数据；
##字典和列表的数据不是一份，而是因为10个进程，所以有10个字典和10个列表。每个进程修改后，都会copy给其他进程，其他进程可以对最新的数据进行修改，所以数据不会被修改乱。

进程同步

在进程里面也有锁

from multiprocessing import Process, Lock   #从multiprocessing导入Lock这个锁

def f(l, i):
    l.acquire()     #获取修改数据的锁
    print(‘hello world‘, i)
    l.release()     #释放锁

if __name__ == ‘__main__‘:
    lock = Lock()   #实例锁

    for num in range(10):   #生成10个进程
        Process(target=f, args=(lock, num)).start() #执行子进程并传入参数给子进程

执行结果：
hello world 1
hello world 4
hello world 0
hello world 3
hello world 2
hello world 5
hello world 6
hello world 8
hello world 7
hello world 9
## 可以看到一共10个进程，并不是连续的，说明执行进程的时候说不准先执行哪个进程。

‘‘‘
进程之间数据是独立的，这里我们为什么又要加锁呢，这是因为所有进程使用同一个屏幕来输出数据；
比如 我们现在输出的数据是 hello world x，在输出的过程中很有可能其中一个进程还没输出完（比如只输出了hello wo），另一个进程就执行输出了（可能会在屏幕上看到hello wohello world0201的现象）。
所以需要通过锁来控制同一时间只能有一个进程输出数据到屏幕。
‘‘‘

进程池

执行多进程，子进程会从主进程复制一份完整数据，1个、10个进程可能还没什么感觉，但是如果有100或1000，甚至更多个进程的时候开销就会特别大，就会明显感觉到多进程执行有卡顿现象。

进程池可以设定同一时间有多少个进程可以在CPU上运行。

from  multiprocessing import Process, Pool
#从multiprocessing导入pool

import time,os

def Foo(i):
    time.sleep(2)
    print("in process",os.getpid()) #打印进程id
    return i + 100

def Bar(arg):
    print(‘-->exec done:‘, arg)

if __name__ == ‘__main__‘:  ##这行代码用途是如果主动执行该代码的.py文件，则该代码下面的代码可以被执行；如果该.py模块被导入到其他模块中，从其他模块执行该.py模块，则该行下面的代码不会被执行。  有些时候可以用这种方式用于测试，在该行代码下面写一些测试代码。。
    pool = Pool(5)  #同时只能放入5个进程

    for i in range(10): #创建10个进程,但是因为pool的限制，只有放入进程池中的5个进程才会被执行（），其他的被挂起了，如果进程池中其中有两个进程执行完了，就会补进2个进程进去。
        # pool.apply_async(func=Foo, args=(i,), callback=Bar)
        pool.apply(func=Foo, args=(i,)) #pool.apply用来将进程放入pool

    print(‘end‘)    #执行完毕
    pool.close()    #允许pool中的进程关闭（close必须在join前面，可以理解close相当于一个开关吧）
    pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

执行结果：
in process 13616
in process 10912
in process 12472
in process 15180
in process 12404
in process 13616
in process 10912
in process 12472
in process 15180
in process 12404
end

##可以看到通过串行的方式将结果打印出来，这是因为我们使用的是pool.apply。 pool.apply就是通过串行的方式来执行。

from  multiprocessing import Process, Pool
import time,os

def Foo(i):
    time.sleep(2)
    print("in process",os.getpid())
    return i + 100

def Bar(arg):
    print(‘-->exec done:‘, arg)

if __name__ == ‘__main__‘:
    pool = Pool(5)

    for i in range(10):
        pool.apply_async(func=Foo, args=(i,))
        ## 使用pool.apply_async就可以并行了

    print(‘end‘)
    pool.close()
    # pool.join()   注释掉

执行结果：
end
## 只执行了print(‘end‘)代码，其他进程的结果没有看到，这是因为其他进程还没有执行完成，主进程pool.close()就执行完了，close以后所有其他进程也不会在执行了。
## 要想其他进程执行完成后在关闭，必须使用pool.join()

from  multiprocessing import Process, Pool
import time,os

def Foo(i):
    time.sleep(2)
    print("in process",os.getpid())
    return i + 100

def Bar(arg):
    print(‘-->exec done:‘, arg)

if __name__ == ‘__main__‘:
    pool = Pool(5)

    for i in range(10):
        pool.apply_async(func=Foo, args=(i,))

    print(‘end‘)
    pool.close()
    pool.join()

执行结果：
end
in process 14756
in process 14596
in process 10836
in process 12536
in process 12904
in process 14756
in process 14596
in process 10836
in process 12536
in process 12904

##从执行结果来看，5个 5个的被打印出来。

回调

from  multiprocessing import Process, Pool
import time,os

def Foo(i):
    time.sleep(2)
    print("in process",os.getpid())
    return i + 100

def Bar(arg):
    print(‘-->exec done:‘, arg,os.getpid())

if __name__ == ‘__main__‘:
    pool = Pool(5)

    print ("主进程：",os.getpid())  #打印主进程id
    for i in range(10):
        pool.apply_async(func=Foo, args=(i,),callback=Bar)
        ##callback叫做回调，就是当执行完了func=Foo后，才会执行callback=Bar(每个进程执行完了后都会执行回调)。
        ## 回调可以用于当执行完代码后做一些后续操作，比如查看完命令后，通过回调进行备份；或者执行完什么动作后，做个日志等。
        ## 备份、写日志等在子进程中也可以执行，但是为什么要用回调呢！ 这是因为如果用子进程，有10个子进程就得连接数据库十次，而使用回调的话是用主进程连接数据库，所以只连接一次就可以了，这样写能大大提高运行效率。
        ##通过主进程建立数据库的连接的话，因为在同一个进程中只能在数据库建立一次连接，所以即使是多次被子进程回调，也不会重复建立连接的，因为数据库会限制同一个进程最大连接数，这都是有数据库设置的。

    print(‘end‘)
    pool.close()
    pool.join()

执行结果：
主进程： 14340    #主进程是 14340
end
in process 13936
-->exec done: 100 14340  #可以看出回调是通过主线程调用的
in process 15348
-->exec done: 101 14340
in process 10160
-->exec done: 102 14340
in process 11612
-->exec done: 103 14340
in process 14836
-->exec done: 104 14340
in process 13936
-->exec done: 105 14340
in process 15348
-->exec done: 106 14340
in process 10160
-->exec done: 107 14340
in process 11612
-->exec done: 108 14340
in process 14836
-->exec done: 109 14340　　

文章根据 代码老兵 的分享博客，一点点搞出来的，多线程和进程让我头疼了三天，感谢大神们的分享的经验，让我少走弯路。