内容简介：python网络编程（进程与多线程）
multiprocessing模块
由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在 python 中大部分情况需要使用多进程。
multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。
此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。
multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。
Process类的介绍
Process(target = talk,args = (conn,addr)) #由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）
group参数未使用，值始终为None，
target表示调用对象，即子进程要执行的任务，
args表示调用对象的位置参数元组，args=(1,2,'egon',)，
kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}，
name为子进程的名称。
方法：p.start()：启动进程，并调用该子进程中的p.run()
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
属性：p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid，每个进程都会开启一个python解释器去完成，对应一个pid号。
from multiprocessing import Process
import os
import time
def info(name):
print("name:",name)
print('parent process:', os.getppid())
print('process id:', os.getpid())
print("------------------")
time.sleep(1)
def foo(name):
info(name)
if __name__ == '__main__':
info('main process line')
p1 = Process(target=info, args=('alvin',))
p2 = Process(target=foo, args=('egon',))
p1.start()
p2.start()
p1.join()
p2.join()
print("ending")
运行结果：
name: main process line
parent process: 7904#pycharm的进程pid
process id: 11424#这个是python解释器的pid
------------------
name: alvin
parent process: 11424
process id: 9628
------------------
name: egon
parent process: 11424
process id: 9276
------------------
ending
pid
p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束。
p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功。
使用方式分为直接调用和继承类方式调用：
from multiprocessing import Process
import time,random
import os
def piao(name):
print(os.getppid(),os.getpid())
print('%s is piaoing' %name)
time.sleep(random.randint(1,3))
print('%s is piao end' %name)
if __name__ == '__main__':
p1=Process(target=piao,kwargs={'name':'alex',})
p2=Process(target=piao,args=('wupeiqi',))
p3=Process(target=piao,kwargs={'name':'yuanhao',})
p1.start()
p2.start()
p3.start()
print('主进程',os.getpid())
#os.getppid(),os.getpid()
#父进程id，当前进程id
开启进程方式一
from multiprocessing import Process
import time,random
import os
class Piao(Process):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
print(os.getppid(),os.getpid())
print('%s is piaoing' %self.name)
# time.sleep(random.randint(1,3))
print('%s is piao end' %self.name)
if __name__ == '__main__':
p1=Piao('alex')
p2=Piao('wupeiqi')
p3=Piao('yuanhao')
p1.start()
p2.start()
p3.start()
print('主进程',os.getpid(),os.getppid())
开启进程方式二
协程函数
协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：
协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。
import time
"""
传统的生产者-消费者模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但一不小心就可能死锁。
如果改用协程，生产者生产消息后，直接通过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高。
"""
# 注意到consumer函数是一个generator（生成器）:
# 任何包含yield关键字的函数都会自动成为生成器(generator)对象
def consumer():
r = ''
while True:
# 3、consumer通过yield拿到消息，处理，又通过yield把结果传回；
# yield指令具有return关键字的作用。然后函数的堆栈会自动冻结(freeze)在这一行。
# 当函数调用者的下一次利用next()或generator.send()或for-in来再次调用该函数时，
# 就会从yield代码的下一行开始，继续执行，再返回下一次迭代结果。通过这种方式，迭代器可以实现无限序列和惰性求值。
n = yield r
if not n:
return
print('[CONSUMER] ←← Consuming %s...' % n)
time.sleep(1)
r = '200 OK'
def produce(c):
# 1、首先调用c.next()启动生成器
next(c)
n = 0
while n < 5:
n = n + 1
print('[PRODUCER] →→ Producing %s...' % n)
# 2、然后，一旦生产了东西，通过c.send(n)切换到consumer执行；
cr = c.send(n)
# 4、produce拿到consumer处理的结果，继续生产下一条消息；
print('[PRODUCER] Consumer return: %s' % cr)
# 5、produce决定不生产了，通过c.close()关闭consumer，整个过程结束。
c.close()
if __name__=='__main__':
# 6、整个流程无锁，由一个线程执行，produce和consumer协作完成任务，所以称为“协程”，而非线程的抢占式多任务。
c = consumer()
produce(c)
'''
result:
[PRODUCER] →→ Producing 1...
[CONSUMER] ←← Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 2...
[CONSUMER] ←← Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 3...
[CONSUMER] ←← Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 4...
[CONSUMER] ←← Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 5...
[CONSUMER] ←← Consuming 5...
[PRODUCER] Consumer return: 200 OK
'''
协程函数
from greenlet import greenlet
def test1():
print (12)
gr2.switch()
print (34)
gr2.switch()
def test2():
print (56)
gr1.switch()
print (78)
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()
greenlet
gevent模块实现协程
Python通过yield提供了对协程的基本支持，但是不完全。而第三方的gevent为Python提供了比较完善的协程支持。
gevent是第三方库，通过greenlet实现协程，其基本思想是：
当一个greenlet遇到IO操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO。
由于切换是在IO操作时自动完成，所以gevent需要修改Python自带的一些标准库，这一过程在启动时通过monkey patch完成：
import gevent
import time
def foo():
print("running in foo")
gevent.sleep(2)
print("switch to foo again")
def bar():
print("switch to bar")
gevent.sleep(5)
print("switch to bar again")
start=time.time()
gevent.joinall(
[gevent.spawn(foo),
gevent.spawn(bar)]
)
print(time.time()-start)
gevent示例
实际代码里，我们不会用gevent.sleep()去切换协程，而是在执行到IO操作时，gevent自动切换，代码如下：
from gevent import monkey
monkey.patch_all()
import gevent
from urllib import request
import time
def f(url):
print('GET: %s' % url)
resp = request.urlopen(url)
data = resp.read()
print('%d bytes received from %s.' % (len(data), url))
start=time.time()
gevent.joinall([
gevent.spawn(f, 'https://itk.org/'),
gevent.spawn(f, 'https://www.github.com/'),
gevent.spawn(f, 'https://zhihu.com/'),
])
# f('https://itk.org/')这是分别爬，串行的操作
# f('https://www.github.com/')
# f('https://zhihu.com/')
print(time.time()-start)
协程在爬网页的I/O
I/O模型
一共有五种类型的I/O模型：1.阻塞I/O：全程阻塞，2.非阻塞I/O：发送多次系统调用，3.IO多路复用（监听多个连接）4.异步IO5.驱动信号
对于一个network IO (这里我们以read举例)，它会涉及到两个系统对象，一个是调用这个IO的process (or thread)，另一个就是系统内核(kernel)。当一个read操作发生时，它会经历两个阶段：
- 等待数据准备 (Waiting for the data to be ready)
- 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)
1.阻塞I/O
在 linux 中，默认情况下所有的socket都是blocking，一个典型的读操作流程大概是这样：
这两个阶段都是阻塞的，在进行的时候不可以做其他的任务，所以是全程阻塞。
non-blocking IO（非阻塞IO）
import socket
import time
sock=socket.socket()
sock.bind(("127.0.0.1",8800))
sock.listen(5)
sock.setblocking(False)#设置为非阻塞
while 1:
try:
conn,addr=sock.accept() # 阻塞等待链接
except Exception as e:
print(e)
time.sleep(3)
非阻塞I/Oserver
import time
import socket
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
while True:
sk.connect(('127.0.0.1',6667))
print("hello")
sk.sendall(bytes("hello","utf8"))
time.sleep(2)
break
client
copy data的时候是阻塞的，等待数据时在监听，数据不来就做其他的事，数据来了就复制数据。
优点：能够在等待任务完成的时间里干其他活了（包括提交其他任务，也就是 “后台” 可以有多个任务在同时执行）。
缺点：任务完成的响应延迟增大了，因为每过一段时间才去轮询一次read操作，而任务可能在两次轮询之间的任意时间完成。这会导致整体数据吞吐量的降低。并且数据也不是实时的，在数据没来时进行某个操作，操作期间数据来了，但是他不能立刻去copy data。
IO multiplexing（IO多路复用）
IO multiplexing就是select，epoll实现的。有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。它的流程如图：
当用户进程调用了select，那么整个进程会被block，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程。
import socket
import time
import select
sock=socket.socket()
sock.bind(("127.0.0.1",8800))
sock.listen(5)
sock.setblocking(False)
inputs=[sock,]
print("sock",sock)
while 1:
r,w,e=select.select(inputs,[],[]) # 监听有变化的套接字 inputs=[sock,conn1,conn2,conn3..]
print("r",r)#select就卡在这一有链接来就开始操作，没链接就block阻塞
print("r",w)
print("r",e)
for obj in r: # 第一次 [sock,] 第二次 #[conn1,]
if obj==sock:#sock是用户来连接我我有变化，将新的链接加入
print('change')
conn,addr=obj.accept()
inputs.append(conn) # inputs=[sock,conn]
else:#客户端传来消息，那么我的conn发生变化，进行数据交互
data=obj.recv(1024)
print(data.decode("utf8"))
send_data=input(">>>")
obj.send(send_data.encode("utf8"))
I/O多路复用并发server
import socket
sock=socket.socket()
sock.connect(("127.0.0.1",8800))
while 1:
data=input("input>>>")
sock.send(data.encode("utf8"))
rece_data=sock.recv(1024)
print(rece_data.decode("utf8"))
sock.close()
client
select仅仅使用I/O多路复用就完成了并发。一开始只监听sock，一有客户端来连接将conn加入监听，然后传数据过来就只监听conn传数据，简单来说select只监听有变化的套接字，没有变化的套接字传输还是按照之前学的套接字之间的数据传输。
结论: select的优势在于可以处理多个连接，不适用于单个连接。
Asynchronous I/O（异步IO）
全程无阻塞，异步就是用户进程发起read操作之后，立刻就可以开始去做其它的事。而另一方面，从kernel的角度，当它受到一个asynchronous read之后，首先它会立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给用户进程发送一个signal，告诉它read操作完成了。
到目前为止，已经将四个IO Model都介绍完了。现在回过头来回答最初的那几个问题：blocking和non-blocking的区别在哪，synchronous IO和asynchronous IO的区别在哪。
调用blocking IO会一直block住对应的进程直到操作完成，而non-blocking IO在kernel还准备数据的情况下会立刻返回。
各个IO Model的比较如图所示：
non-blocking IO中，虽然进程大部分时间都不会被block，但是它仍然要求进程去主动的check，并且当数据准备完成以后，也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人（kernel）完成，然后他人做完后发信号通知。在此期间，用户进程不需要去检查IO操作的状态，也不需要主动的去拷贝数据。
selectors模块（基于select机制实现的IO多路复用）
这个模块已经封装了select，poll，和epoll实现I/O多路复用。
windows下只有select，linux上还有poll和epoll。
select缺点每次调用都要将所有文件描述符copy到内核空间导致效率低，每次都要遍历所有的fd，是否有数据访问。最大连接数1024，poll只是没有连接数限制。
epoll：第一个函数创建epoll句柄，只有第一次要将所有文件描述符copy到内核空间，第二个函数回调函数，某一个函数某一个动作成功完成后会触发的函数，为所有fd绑定回调函数，一旦有数据访问触发此回调函数，回调函数将fd放到链表中。第三个函数判断链表是否为空。
import selectors # 基于select模块实现的IO多路复用，建议大家使用
import socket
sock=socket.socket()
sock.bind(("127.0.0.1",8800))
sock.listen(5)
sock.setblocking(False)
sel=selectors.DefaultSelector() #根据具体平台选择最佳IO多路机制，比如在linux，选择epoll
def read(conn,mask):
try:
data=conn.recv(1024)
print(data.decode("UTF8"))
data2=input(">>>")
conn.send(data2.encode("utf8"))
except Exception:
sel.unregister(conn)
def accept(sock,mask):
conn, addr = sock.accept()
print("conn",conn)
sel.register(conn,selectors.EVENT_READ,read)
sel.register(sock,selectors.EVENT_READ,accept) # selectors对象注册事件，监听谁就要注册谁，第二个默认，第三个监听对象有变化运行这个函数
while 1:
print("wating...")
events=sel.select() # 监听 [(key1,mask1),(key2,mask2)]第一次只监听sock有链接过来才会继续
for key,mask in events:
# print(key.fileobj) # conn
# print(key.data) # read
func=key.data#accept函数
obj=key.fileobj#sock
func(obj,mask) # 1 accept(sock,mask) # 2 read(conn,mask)
selectors模块server
很明显封装好的模块省去了select的底层操作，用起来简便很多。
队列
多线程多进程才有队列的概念。队列是数据类型。
import queue
#q=queue.Queue(3) # 默认是 先进先出(FIFO)管道容纳最大值
# q.put(111)#put塞值
# q.put("hello")
# q.put(222)
#
# q.put(223,False)
#
# print(q.get())#get取值取不到值就是block
# # print(q.get())
# # print(q.get())
# #
# q.get(False)#取不到还把block转成false就会报错了。
# queue 优点： 线程安全的
queue
q=Queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数 maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。
q.put(10) 调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目值； 第二个block为可选参数，默认为 1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0， put方法将引发Full异常。
将一个值从队列中取出 q.get() 调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且 block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。
# join和task_done
# q=queue.Queue(5)
# q.put(111)
# q.put(222)
# q.put(22222)
#
#
# while not q.empty():
# a=q.get()
# print(a)
#q.task_done()#任务完成了告诉一下join
# b=q.get()
# print(b)
# q.task_done()
# q.join()只有所有的都任务都结束才不block否则都卡住
#
# print("ending")
join，task_done
join() 阻塞进程，直到所有任务完成，需要配合另一个方法task_done。
task_done() 表示某个任务完成。每一条get语句后需要一条task_done。
其他常用方法
此包中的常用方法(q = Queue.Queue()):
q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)非阻塞
q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当
q.put(item, False)
q.task_done() 在完成一项工作之后，
q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作。
其他模式
Python Queue模块有三种队列及构造函数:
1、Python Queue模块的FIFO队列先进先出。 class queue.Queue(maxsize)
2、LIFO类似于堆，即先进后出。 class queue.LifoQueue(maxsize)
3、还有一种是优先级队列级别越低越先出来。 class queue.PriorityQueue(maxsize )
# 先进后出模式 # q=queue.LifoQueue() # Lifo last in first out # # # q.put(111) # q.put(222) # q.put(333) # # print(q.get()) 先进后出模式
# 优先级 # q=queue.PriorityQueue() # # q.put([4,"hello4"]) # q.put([1,"hello"]) # q.put([2,"hello2"]) # # print(q.get()) # print(q.get()) 优先级
生产者消费者模型
是一种设计模式。我们来模拟这个过程。
#生产者消费者模型
import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
count = 0
while count <10:#生产数据
print("making........")
time.sleep(2)#生产时间
q.put(count)
print('Producer %s has produced %s baozi..' %(name, count))
count +=1
#q.task_done()
#q.join()
print("ok......")
def Consumer(name):#消费者
count = 0
while count <10:
time.sleep(1)
if not q.empty():
data = q.get()
#q.task_done()
#q.join()
print(data)
print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))
else:
print("-----no baozi anymore----")
count +=1
p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
生产者消费者模型
以上所述就是小编给大家介绍的《python网络编程（进程与多线程）》，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持！
猜你喜欢:
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
Out of their Minds
Dennis Shasha、Cathy Lazere / Springer / 1998-07-02 / USD 16.00
This best-selling book is now available in an inexpensive softcover format. Imagine living during the Renaissance and being able to interview that eras greatest scientists about their inspirations, di......一起来看看 《Out of their Minds》 这本书的介绍吧!
