本节内容
- 多进程multiprocessing
- 进程间的通讯
- 协程
- 论事件驱动与异步IO
- SelectPollEpoll——IO多路复用
1、多进程multiprocessing
Python的线程用的是操作系统的原生线程，同样python的进程用的是操作系统的原生进程。
多进程之间没有锁的概念，多进程之间数据不能互相访问，所以不存在互斥锁。GIL问题又是仅仅出现在多线程中。
所以如果我们启动8个进程，每个进程有一个主线程，即8个线程，分别运行在8个CPU上，就可以充分利用多核的优势了。
在多进程充分利用多核的优势下，唯一的坏处是这8个进程之间数据无法共享。传递数据需要找媒介。
8进程表示，同一时间最多只能干8件事情。
所以多进程可以解决多核的问题
同时每个进程里又可以写多个线程，启动并执行。
每一个进程都是由它的父进程启动的。
os.getppid() #获得父进程的id
os.getpid() #获得自己进程的id
|
1
2
3
4
5
6
7
8
9
10
|
from
multiprocessing
import
Process
import
time
def
f(name):
time.sleep(
2
)
print
(
'hello'
, name)
if
__name__
=
=
'__main__'
:
p
=
Process(target
=
f, args
=
(
'bob'
,))
p.start()
p.join()
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
from
multiprocessing
import
Process
import
os
def
info(title):
print
(title)
print
(
'module name:'
, __name__)
print
(
'parent process:'
, os.getppid())
print
(
'process id:'
, os.getpid())
print
(
"nn"
)
def
f(name):
info(
' 33[31;1mfunction f 33[0m'
)
print
(
'hello'
, name)
if
__name__
=
=
'__main__'
:
info(
' 33[32;1mmain process line 33[0m'
)
p
=
Process(target
=
f, args
=
(
'bob'
,))
p.start()
p.join()
|
2、进程间的通讯
不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以用以下方法：
Queues（进程队列）——数据传递
使用方法跟threading里的queue差不多，
原理上是两个进程各有一个queue队列，通过pickle序列化的方式，实现两个队列之间的交互，从而看起来像两个进程之间的交互。
所以实际上进程队列是2个Queues，而不是一个共享queue队列。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
from
multiprocessing
import
Process, Queue
def
f(q):
q.put([
42
,
None
,
'hello'
])
if
__name__
=
=
'__main__'
:
q
=
Queue()
p
=
Process(target
=
f, args
=
(q,))
p.start()
print
(q.get())
# prints "[42, None, 'hello']"
p.join()
|
Pipes（管道）——数据传递
pipe管道就相当于拿了一根电话线，两头分别连接了两个进程。原理是利用socket网络协议进行周转。
pipe实例一生成会产生两个返回对象，一个是管道的一头，另一个是管道的另一头。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
from
multiprocessing
import
Process, Pipe
def
f(conn):
conn.send([
42
,
None
,
'hello'
])
conn.close()
if
__name__
=
=
'__main__'
:
parent_conn, child_conn
=
Pipe()
p
=
Process(target
=
f, args
=
(child_conn,))
p.start()
print
(parent_conn.recv())
# prints "[42, None, 'hello']"
p.join()
|
Managers已经自动加锁，所以不需要手动加锁。
A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.
A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example,
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
from
multiprocessing
import
Process, Manager
def
f(d, l):
d[
1
]
=
'1'
d[
'2'
]
=
2
d[
0.25
]
=
None
l.append(
1
)
print
(l)
if
__name__
=
=
'__main__'
:
with Manager() as manager:
d
=
manager.
dict
()
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)
|
进程锁
虽然本身由于进程间不能共享数据，所以不需要锁。但是当进程往屏幕上打印数据时，对各个进程而言屏幕是共享的。所以这个锁的目的是锁住一个时间只能一个进程去打印数据。保证打印数据的完整性，前面正在打印的数据，不被后面要打印的数据打断。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
from
multiprocessing
import
Process, Lock
def
f(l, i):
l.acquire()
try
:
print
(
'hello world'
, i)
finally
:
l.release()
if
__name__
=
=
'__main__'
:
lock
=
Lock()
for
num
in
range
(
10
):
Process(target
=
f, args
=
(lock, num)).start()
|
进程池
进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。
进程池中有两个方法：
- apply #进程池串行
- apply_async #进程池并行
对于进程池中进程的启动是要使用pool.apply()或pool.apply_async()
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from
multiprocessing
import
Process,Pool,freeze_support
import
time
def
Foo(i):
time.sleep(
2
)
return
i
+
100
def
Bar(arg):
print
(
'-->exec done:'
,arg)
if __name__ == '__main__':
freeze_support() #在windows上必须添加这行和上行代码
pool
=
Pool(
5
) #允许进程池里同时放入5个进程
for
i
in
range
(
10
):
pool.apply_async(func
=
Foo, args
=
(i,),callback
=
Bar)
#
pool.apply_async(func
=
Foo, args
=
(i,)
)
#pool.apply(func=Foo, args=(i,))
print
(
'end'
)
pool.close()
pool.join()
#进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。
|
这里不知何种原因，必须是先pool.close()，然后再是pool.join()
如果最后不加pool.join()，主程序不会等待进程池执行完毕，会直接关闭。
if __name__ == '__main__'： #这句话是为了区分，主动启动脚本还是把它当成一个模块，从别的地方去调用。
如果主动启动该脚本，则该段代码下面部分代码会被执行。如果从别的地方调用，则下面代码不会被执行。
主动执行该py文件时，把这句话当成主程序的入口。
上面代码中含有pool.apply_async(func=Foo, args=(i,),callback=Bar)，callback叫做回调，意思是执行完这个语句后再回调Bar函数。这里需要注意，callback方法的调用进程是主进程，而不是子进程。
回调函数的意义，如果开了100个子进程，在每个子进程结束时往数据库中插入数据，需要建立100个连接，但如果先把100个子进程的结果保存到变量里，让父进程连接数据库一次，然后一口气全部插入，就大大提高了程序的运行速度。此时的回调函数，应运而生。
3、协程
协程，又称微线程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程。
CPU只认识线程，并不知道协程的存在。协程是跑在线程中的。
协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：
协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。
换句话说：在单线程下，实现并发的效果，就是协程。例如，之前使用yield做的生产者消费者模型。
协程的好处：
- 无需线程上下文切换的开销
- 无需原子操作锁定及同步的开销
- "原子操作(atomic operation)是不需要synchronized"，所谓原子操作是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另一个线程）。原子操作可以是一个步骤，也可以是多个操作步骤，但是其顺序是不可以被打乱，或者切割掉只执行部分。视作整体是原子性的核心。
- 方便切换控制流，简化编程模型
- 高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。
缺点：
- 无法利用多核资源：协程的本质是个单线程,它不能同时将单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
- 进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序
使用yield实现协程操作的例子
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
import
time
import
queue
def
consumer(name):
print
(
"--->starting eating baozi..."
)
while
True
:
new_baozi
=
yield #yield这里本身可以返回数据，也可接受数据
print
(
"[%s] is eating baozi %s"
%
(name,new_baozi))
#time.sleep(1)
def
producer():
r
=
con.__next__()
r
=
con2.__next__()
n
=
0
while
n <
5
:
n
+
=
1
con.send(n) #激活yield的阻塞状态，同时传入一个数据
con2.send(n)
print
(
" 33[32;1m[producer] 33[0m is making baozi %s"
%
n )
if
__name__
=
=
'__main__'
:
con
=
consumer(
"c1"
)
con2
=
consumer(
"c2"
)
p
=
producer()
|
协程之所以能处理大并发，就是把I/O操作给挤掉了，即一旦遇到I/O操作就切换。使得整个程序变成了只有CPU运算，大大提高了效率。
只要I/O操作一完成，CPU就可以切换回去了。这样就把I/O操作完全挤出去了。程序会自动检测I/O是否完成，不需要我们关心。
协程的标准定义，即符合什么条件就能称之为协程：
- 必须在只有一个单线程里实现并发
- 修改共享数据不需加锁
- 用户程序里自己保存多个控制流的上下文栈
- 一个协程遇到IO操作自动切换到其它协程
Greenlet
yield是自己写的协程，greenlet是一个封装好的协程。
greenlet是一个用C实现的协程模块，相比与python自带的yield，它可以使你在任意函数之间随意切换，而不需把这个函数先声明为generator
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
# -*- coding:utf-8 -*-
from
greenlet
import
greenlet
def
test1():
print
(
12
)
gr2.switch() #gr2.switch()的意思是切换到gr2
print
(
34
)
gr2.switch() #同上
def
test2():
print
(
56
)
gr1.switch()
print
(
78
)
gr1
=
greenlet(test1) #启动一个协程
gr2
=
greenlet(test2)
gr1.switch() #gr1.switch()的意思是切换到gr1
|
感觉确实用着比generator
原文链接：https://blog.csdn.net/weixin_33908217/article/details/94541395
站方申明：本站部分内容来自社区用户分享，若涉及侵权，请联系站方删除。
