python并发编程补充
一、进程通信1、信号量
互斥锁：同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据
如果指定信号量为3，那么来一个人获得一把锁，计数加1，当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁，信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念。
from multiprocessing import Process,Semaphore
import time,random
def action(sem,user):
sem.acquire()
print('%s 占一个位置' %user)
time.sleep(random.randint(0,3)) #模拟进程执行时间
sem.release()
if __name__ == '__main__':
sem=Semaphore(5)
p_l=[] #存放启动的进程
for i in range(13):
p=Process(target=action,args=(sem,'user%s' %i,))
p.start()
p_l.append(p)
for i in p_l:
i.join() #保证所以子进程执行完毕
print('============》')
2、事件
python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。
事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
clear：将“Flag”设置为False
set：将“Flag”设置为True
3、进程池
在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：
（1）很明显需要并发执行的任务通常要远大于核数
（2）一个操作系统不可能无限开启进程，通常有几个核就开几个进程
（3）进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）
我们就可以通过维护一个进程池来控制进程数目
对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。
创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程.
4、使用进程池维护固定数目的进程
# 开启6个客户端，会发现2个客户端处于等待状态
# 服务端
from socket import *
from multiprocessing import Pool
import os
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)
def talk(conn,client_addr):
print('进程pid: %s' %os.getpid())
while True:
try:
msg=conn.recv(1024)
if not msg:break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
p=Pool() #Pool内的进程数默认是cpu核数，假设为4（查看方法os.cpu_count()）
while True:
conn,client_addr=server.accept()
p.apply_async(talk,args=(conn,client_addr))
# p.apply(talk,args=(conn,client_addr)) #同步的话，则同一时间只有一个客户端能访问
# 客户端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
msg=input('>>: ').strip()
if not msg:continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))
5、回调函数
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。
# 爬虫案例
from multiprocessing import Pool
import time,random
import requests
import re
def get_page(url,pattern):
response=requests.get(url)
if response.status_code == 200:
return (response.text,pattern)
def parse_page(info):
page_content,pattern=info
res=re.findall(pattern,page_content)
for item in res:
dic={
'index':item,
'title':item,
'actor':item.strip(),
'time':item,
'score':item+item
}
print(dic)
if __name__ == '__main__':
pattern1=re.compile(r'<dd>.*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S)
url_dic={
'http://maoyan.com/board/7':pattern1,
}
p=Pool()
res_l=[]
for url,pattern in url_dic.items():
res=p.apply_async(get_page,args=(url,pattern),callback=parse_page)
res_l.append(res)
for i in res_l:
i.get()
# res=requests.get('http://maoyan.com/board/7')
# print(re.findall(pattern,res.text))
'''结果：
{'index': '1', 'title': '神秘巨星', 'actor': '阿米尔·汗,塞伊拉·沃西,梅·维贾', 'time': '2018-01-19', 'score': '9.5'}
{'index': '2', 'title': '奇迹男孩', 'actor': '雅各布·特瑞布雷,朱莉娅·罗伯茨,欧文·威尔逊', 'time': '2018-01-19', 'score': '9.3'}
{'index': '3', 'title': '小狗奶瓶', 'actor': '奶瓶,康潇诺,魏子涵', 'time': '2018-02-02', 'score': '9.3'}
{'index': '4', 'title': '公牛历险记', 'actor': '约翰·塞纳,莉莉·戴,凯特·迈克金农', 'time': '2018-01-19', 'score': '9.2'}
{'index': '5', 'title': '前任3：再见前任', 'actor': '韩庚,郑恺,于文文', 'time': '2017-12-29', 'score': '9.2'}
{'index': '6', 'title': '一个人的课堂', 'actor': '孙海英,韩三明,王乃训', 'time': '2018-01-16', 'score': '9.2'}
{'index': '7', 'title': '芳华', 'actor': '黄轩,苗苗,钟楚曦', 'time': '2017-12-15', 'score': '9.1'}
{'index': '8', 'title': '南极之恋', 'actor': '赵又廷,杨子姗', 'time': '2018-02-01', 'score': '9.0'}
{'index': '9', 'title': '马戏之王', 'actor': '休·杰克曼,扎克·埃夫隆,米歇尔·威廉姆斯', 'time': '2018-02-01', 'score': '9.0'}
{'index': '10', 'title': '小马宝莉大电影', 'actor': '奥卓·阿杜巴,艾米莉·布朗特,克里斯汀·肯诺恩斯', 'time': '2018-02-02', 'score': '8.9'}
'''
6、如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数
from multiprocessing import Pool
import time,random,os
def work(n):
time.sleep(1)
return n**2
if __name__ == '__main__':
p=Pool()
res_l=[]
for i in range(10):
res=p.apply_async(work,args=(i,))
res_l.append(res)
p.close()
p.join() #等待进程池中所有进程执行完毕
nums=[]
for res in res_l:
nums.append(res.get()) #拿到所有结果
print(nums) #主进程拿到所有的处理结果,可以在主进程中进行统一进行处理
二、通信线程
1、死锁现象与递归锁
所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程
解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。
from threading import Thread,RLock
import time
mutexA=mutexB=RLock()
class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print('\033[41m%s 拿到A锁\033[0m' %self.name)
mutexB.acquire()
print('\033[42m%s 拿到B锁\033[0m' %self.name)
mutexB.release()
mutexA.release()
def func2(self):
mutexB.acquire()
print('\033[43m%s 拿到B锁\033[0m' %self.name)
time.sleep(2)
mutexA.acquire()
print('\033[44m%s 拿到A锁\033[0m' %self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t=MyThread()
t.start()
2、信号量Semaphore
同进程的一样
Semaphore管理一个内置的计数器，每当调用acquire()时内置计数器-1；调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
from threading import Thread,Semaphore,current_thread
import threading
import time,random
def func():
with sm:
print('%s get sm' %current_thread().getName())
time.sleep(random.randint(1,3))
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(20):
t=Thread(target=func)
t.start()
3、Event
同进程的一样
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。
所以我们需要使用threading库中的Event对象。对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在初始情况下,Event对象中的信号标志被
为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,
它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。
event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。
# 模拟连接mysql
from threading import Thread,Event
import threading
import time,random
def conn_mysql():
count=1
while not event.is_set():
if count > 3: #超过次数就抛出异常链接超时
raise TimeoutError('链接超时')
print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count))
event.wait(1) #等待1秒后接着尝试连接
count+=1
print('<%s>链接成功' %threading.current_thread().getName())
def check_mysql():
print('\033正在检查mysql\033[0m' % threading.current_thread().getName())
time.sleep(random.randint(2,4))
event.set()
if __name__ == '__main__':
event=Event()
conn1=Thread(target=conn_mysql)
conn2=Thread(target=conn_mysql)
check=Thread(target=check_mysql)
conn1.start()
conn2.start()
check.start()
4、定时器
# 验证码定时器
from threading import Timer
import random,time
class Code:
def __init__(self):
self.make_cache()
def make_cache(self,interval=5):
self.cache=self.make_code()
print(self.cache)
self.t=Timer(interval,self.make_cache)
self.t.start()
def make_code(self,n=4):
res=''
for i in range(n):
s1=str(random.randint(0,9))
s2=chr(random.randint(65,90))
res+=random.choice()
return res
def check(self):
while True:
inp=input('>>: ').strip()
if inp.upper() ==self.cache:
print('验证成功',end='\n')
self.t.cancel()
break
if __name__ == '__main__':
obj=Code()
obj.check()
5、线程queue
（1）先进先出
import queue
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
print(q.get()) #first
print(q.get()) #second
print(q.get()) #third
（2）堆栈（后进先出）
import queue
q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')
print(q.get()) #third
print(q.get()) #second
print(q.get()) #first
（3）按照优先级取值
import queue
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((20,'a'))
q.put((-5,'b'))
q.put((30,'c'))
print(q.get()) #(-5, 'b')
print(q.get()) #(20, 'a')
print(q.get()) #(30, 'c')
三、Python标准模块--concurrent.futures
1、介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
2、ProcessPoolExecutor用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
print('%s is runing' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor(max_workers=3)
futures=[]
for i in range(11):
future=executor.submit(task,i) #异步提交任务
futures.append(future)
executor.shutdown(True) #wait=True，等待池内所有任务执行完毕回收完资源后才继续，wait=False，立即返回，并不会等待池内的任务执行完毕
print('+++>')
for future in futures:
print(future.result()) #取得结果
3、map的用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
print('%s is runing' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ThreadPoolExecutor(max_workers=3)
# for i in range(11):
# future=executor.submit(task,i)
executor.map(task,range(1,10)) #map取代for循环submit的操作
4、回调函数
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os
def get_page(url):
print('<进程%s> get %s' %(os.getpid(),url))
respone=requests.get(url)
if respone.status_code == 200:
return {'url':url,'text':respone.text}
def parse_page(res):
res=res.result()
print('<进程%s> parse %s' %(os.getpid(),res['url']))
parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))
with open('db.txt','a') as f:
f.write(parse_res)
if __name__ == '__main__':
urls=[
'https://www.baidu.com',
'https://www.python.org',
'https://www.openstack.org',
'https://help.github.com/',
'http://www.sina.com.cn/'
]
# p=Pool(3)
# for url in urls:
# p.apply_async(get_page,args=(url,),callback=pasrse_page)
# p.close()
# p.join()
p=ProcessPoolExecutor(3)
for url in urls:
p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果
5、同步调用和异步调用
提交任务的两种方式：
同步调用：提交完任务后，就在原地等待，等待任务执行完毕，拿到任务的返回值，才能继续下一行代码，导致程序串行执行
异步调用+回调机制：提交完任务后，不在原地等待，任务一旦执行完毕就会触发回调函数的执行， 程序是并发执行
#同步调用示例：
from multiprocessing import Pool
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time,random,os
def task(n):
print('%s is ruuning' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
def handle(res):
print('handle res %s' %res)
if __name__ == '__main__':
pool=ProcessPoolExecutor(2)
for i in range(5):
res=pool.submit(task,i).result()
handle(res)
pool.shutdown(wait=True)
# pool.submit(task,33333)
print('主')
# 异步调用示例：
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time,random,os
def task(n):
print('%s is ruuning' %os.getpid())
time.sleep(random.randint(1,3))
# res=n**2
# handle(res)
return n**2
def handle(res):
res=res.result()
print('handle res %s' %res)
if __name__ == '__main__':
pool=ProcessPoolExecutor(2)
for i in range(5):
obj=pool.submit(task,i)
obj.add_done_callback(handle)
pool.shutdown(wait=True)
print('主')
四、协程
1、原理
基于单线程来实现并发，即只用一个主线程（很明显可利用的cpu只有一个）情况下实现并发，这就要用到协程
对于单线程下，我们不可避免程序中出现io操作，但如果我们能在自己的程序中控制单线程下的多个任务能在一个任务遇到io阻塞时就切换到另外一个任务去计算，这样就保证了该线程能够最大限度地处于就绪态，即随时都可以被cpu执行的状态，相当于我们在用户程序级别将自己的io操作最大限度地隐藏起来，从而可以迷惑操作系统，让其以为该线程好像是一直在计算，io比较少，从而更多的将cpu的执行权限分配给我们的线程，提高程序的运行效率。
协程的本质就是在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，以此来提升效率。
所以需要同时满足以下条件的解决方案：
（1）可以控制多个任务之间的切换，切换之前将任务的状态保存下来，以便重新运行时，可以基于暂停的位置继续执行。
（2）作为1的补充：可以检测io操作，在遇到io操作的情况下才发生切换
2、介绍
协程：是单线程下的并发，又称微线程，纤程。协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的
对比操作系统控制线程的切换，用户在单线程内控制协程的切换有什么优缺点？
优点如下：
#1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
#2. 单线程内就可以实现并发的效果，最大限度地利用cpu
缺点如下：
#1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
#2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程
3、总结协程特点
（1）必须在只有一个单线程里实现并发
（2）修改共享数据不需加锁
（3）用户程序里自己保存多个控制流的上下文栈
（4）一个协程遇到IO操作自动切换到其它协程（如何实现检测IO，就用到了gevent模块（select机制））
4、greenlet模块
#pip3 install greenlet #安装greenlet模块
from greenlet import greenlet
import time
def eat(name):
print('%s eat 1' %name)
time.sleep(100)
g2.switch('wang')
print('%s eat 2' %name)
g2.switch()
def play(name):
print('%s play 1' % name)
g1.switch()
print('%s play 2' % name)
g1=greenlet(eat)
g2=greenlet(play)
g1.switch('wang')
使用greenlet模块可以非常简单地实现多个任务的切换，当切到一个任务执行时如果遇到io，那就原地阻塞，这仍然没有解决遇到IO自动切换来提升效率的问题
5、Gevent模块(遇到IO阻塞时会自动切换任务)
#pip3 install gevent #安装Gevent模块
（1）用法
g1=gevent.spawn(func,1,,2,3,x=4,y=5) #创建一个协程对象g1，spawn括号内第一个参数是函数名，后面可以有多个位置实参或关键字实参，都是传给函数的
g2=gevent.spawn(func2)
g1.join() #等待g1结束
g2.join() #等待g2结束
#或者上述两步合作一步：gevent.joinall()
g1.value #拿到func1的返回值
（2）遇到IO阻塞时会自动切换任务
from gevent import monkey;monkey.patch_all()
import gevent
import time
def eat(name):
print('%s eat 1' %name)
# gevent.sleep(3)
time.sleep(3)
print('%s eat 2' %name)
def play(name):
print('%s play 1' % name)
# gevent.sleep(2)
time.sleep(3)
print('%s play 2' % name)
g1=gevent.spawn(eat,'wang')
g2=gevent.spawn(play,'li')
# gevent.sleep(1)
# g1.join()
# g2.join()
gevent.joinall()
我们可以用threading.current_thread().getName()来查看每个g1和g2，查看的结果为DummyThread-n，即假线程
（3）爬虫
from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time
def get_page(url):
print('GET: %s' %url)
response=requests.get(url)
if response.status_code == 200:
print('%d bytes received from %s' %(len(response.text),url))#统计内容的长度
start_time=time.time()
gevent.joinall([
gevent.spawn(get_page,'https://www.python.org/'),
gevent.spawn(get_page,'https://www.yahoo.com/'),
gevent.spawn(get_page,'https://github.com/'),
])
stop_time=time.time()
print('run time is %s' %(stop_time-start_time))
（4）通过gevent实现单线程下的socket并发（from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前，否则gevent无法识别socket的阻塞）
# 服务端
from gevent import monkey;monkey.patch_all()
from socket import *
import gevent
#如果不想用money.patch_all()打补丁,可以用gevent自带的socket
# from gevent import socket
# s=socket.socket()
def server(server_ip,port):
s=socket(AF_INET,SOCK_STREAM)
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind((server_ip,port))
s.listen(5)
while True:
conn,addr=s.accept()
gevent.spawn(talk,conn,addr)
def talk(conn,addr):
try:
while True:
res=conn.recv(1024)
print('client %s:%s msg: %s' %(addr,addr,res))
conn.send(res.upper())
except Exception as e:
print(e)
finally:
conn.close()
if __name__ == '__main__':
server('127.0.0.1',8080)
# 客户端
#_*_coding:utf-8_*_
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
msg=input('>>: ').strip()
if not msg:continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))
# 多线程并发多个客户端
from threading import Thread
from socket import *
import threading
def client(server_ip,port):
c=socket(AF_INET,SOCK_STREAM) #套接字对象一定要加到函数内，即局部名称空间内，放在函数外则被所有线程共享，那么客户端端口永远一样了
c.connect((server_ip,port))
count=0
while True:
c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
msg=c.recv(1024)
print(msg.decode('utf-8'))
count+=1
if __name__ == '__main__':
for i in range(500):
t=Thread(target=client,args=('127.0.0.1',8080))
t.start()
页:
[1]