进程协程缓存

Posted 2020-08-09 台灯不太亮

 

一、进程： （CPU密集型工作多线程有用）

• 进程创建（开销比较大）：

from multiprocessing import Process

import threading

import time

def function(arg):

time.sleep(2)

print(arg)

if __name__ == ‘__main__‘: # 进程创建的本质是调用os.fork，windows下不支持，只能写在 __name__ == ‘__main__‘ 中

for i in range(10):

p = Process(target=function, args=(i,))

p.start()

# p.join(1)

• 可用方法：
• p.daemon
• p.start()
• p.join()  # 子进程执行完毕后再关闭主进程

二、进程间数据共享

• 每个进程独自维护一份自己的数据，不同进程数据默认不会共享

from multiprocessing import Process

lst = []

def function(arg):

    lst.append(arg) # 每个进程都在该列表的末尾追加上自己的ID

    print(lst)

if __name__ == ‘__main__‘:

    for i in range(10):

        p = Process(target=function, args=(i,))

        p.start()

# 此时，每个进程输出的列表只有一个元素，就是自身的id， [1] [2] [3] ...

• 多进程队列：

from multiprocessing import queues # 多进程模块中一种特殊的队列，可以让进程间数据共享

from multiprocessing import Process

import multiprocessing

def function(arg,p):

    p.put(arg)

    size = p.qsize()

    print(size)

if __name__ == ‘__main__‘:

    que = queues.Queue(20, ctx=multiprocessing)

    for i in range(10):

        p = Process(target=function, args=(i, que))

        p.start()

• 多进程数组：
• 解释：
• 数组中所有元素类型必须一致，且创建时大小已经指定
• 创建数组，则需要指定长度和类型，即大小空间都可以知道，所以在内存中一定是挨着的连续空间
• python中的list，可以存放各种类型数据，实际是用链表实现的，即所有元素在内存中不一定挨着（下一个数据记录上一个数据的位置）
• 局限性：
• 数据类型和数量已经规定死了
• 数组数据类型对应表

----------------------------------------------------------------------------------

‘c‘: ctypes.c_char,  ‘u‘: ctypes.c_wchar,

‘b‘: ctypes.c_byte,  ‘B‘: ctypes.c_ubyte,

‘h‘: ctypes.c_short, ‘H‘: ctypes.c_ushort,

‘i‘: ctypes.c_int,   ‘I‘: ctypes.c_uint,

‘l‘: ctypes.c_long,  ‘L‘: ctypes.c_ulong,

‘f‘: ctypes.c_float, ‘d‘: ctypes.c_double

---------------------------------------------------------------------------------

from multiprocessing import Process

from multiprocessing import Array

array_lst = Array(‘i‘ , [ 1,2,3,4,5])  # 创建一个指定类型和指定大小的数组，需要共享的数据放入数组中

def function(index, lst):

lst[index] = i+10  # 每个进程对数组进行修改

for item in lst:

print(item)

print(‘--------------‘)

if __name__ == ‘__main__‘:

for i in range(5):

p = Process(target=function, args=(i, array_lst))

p.start()

• 多进程字典：

from multiprocessing import Process

from multiprocessing import Manager

def function(arg, d):

key = ‘k‘ + str(arg)

value = ‘v‘ + str(arg)

d[key] = value

print(d.keys(), d.values()) # 返回值均为key或value组织成的列表

if __name__ == ‘__main__‘:

obj = Manager()

dic = obj.dict()  # 创建个特殊类型的字典

for i in range(10):

p = Process(target=function, args=(i, dic))

p.start()

p.join()  # 子进程执行完毕再关闭主进程

# 注意，字典是在主进程中创建的，各个子进程需要修改主进程中的数据

# 进程间通讯是要创建链接的，类似socket通讯，

# 若主进程运行结束，则所有连接也会断掉，所以需要join等待，否则报错

三、进程锁：

• # 注意，其中from multiprocessing import Rlock Lock, Event, Condition, Semaphore 都与线程一样
• example：多个进程修改同一份数据

from multiprocessing import Process

from multiprocessing import RLock

from multiprocessing import Array

import time

def function(idx, lst, lck):

lck.acquire() # 加上进程锁，则即使都卡在这里，将“修改和打印”定为原子操作，则每次打印只改变一个元素

lst[idx] += 10

time.sleep(1)  # 若每个进程执行时间稍长些，则所有进程卡在这里，最后print全为[11,12,13]

for item in lst:

print(item)

print(‘--------------‘)

lck.release()

if __name__ == ‘__main__‘:

lock = RLock()

array_lst = Array(‘i‘, [1, 2, 3])

for i in range(3):

p = Process(target=function, args=(i, array_lst, lock))

p.start()

四、进程池：

• 每次去进程池中获取一个进程，然后去执行
• 若进程池中无空闲进程，则会等待，提供了apply，apply_async方法

from multiprocessing import Pool

import time

def function(arg):

print(arg)

time.sleep(1)

if __name__ == ‘__main__‘:

pool = Pool(3)   # 定义一个进程池，容量为3个进程

for i in range(30): # 启动30个任务，调用30次进程去执行

# pool.apply(func=function, args=(i,))  # 进程串行获取，必须等上个进程任务执行完毕才会去获取新的进程

pool.apply_async(func=function, args=(i,)) # 进程并行获取， 每个进程执行指定任务

# time.sleep(2)  # 与下面搭配，表示子进程运行2秒后，直接终止掉，则只会有一部分进程执行完了任务

# pool.terminate() # 立即终止所有子进程，

pool.close() # 等待所有任务执行完毕

pool.join() # 等待close或terminate断言条件满足，再继续往后走，若没有断言，则会抛出异常

print(‘end‘)

五、协程

• 协程：
• 线程和进程的操作是程序处罚系统接口，最后执行者是系统，本质是操作系统提供的并发功能
• 协程是由程序员指定的，人为的实现并发处理

• 存在意义：
• 对于多线程应用，CPU通过上下文切换来实现线程切换，过程比较消耗实际那
• 协程则只需要使用一个线程，分解一个线程为多个“微线程”，在一个线程内部再次实现并发
• 适用场景：
• 程序中存在大量IO操作
• gevent：
• 是对greenlet的高级封装，因此一般用他，
• 通过joinall将任务统一调度，实现单线程中并发微线程

 

import gevent

import time

from gevent import monkey;monkey.patch_all() # 对原先模块打补丁（不修改源码的情况下通过新增类实现）

import requests

def function(url):

resp = requests.get(url)

time.sleep(1)

print(url, len(resp.content))

gevent.joinall(

[

gevent.spawn(function, ‘http://www.baidu.com‘), # 指定要执行的任务，和任务的参数

gevent.spawn(function, ‘http://www.hao123.com‘),

gevent.spawn(function, ‘http://www.qq.com‘),

]

)