十三、并发编程之多线程
理论：http://www.cnblogs.com/linhaifeng/articles/7430082.html
连接：http://www.cnblogs.com/linhaifeng/articles/7428877.html
1、线程
1.1 什么是线程
进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位。
多线程（即多个控制线程）的概念是，在一个进程中存在多个控制线程，多个控制线程共享该进程的地址空间，相当于一个车间内有多条流水线，都共用一个车间的资源。
1.2 特点
#1 同一进程内的多个线程是共享该进程的资源
#2 创建新的线程开销要远远小于开启新的进程
1.3 开启线程的方式
1.3.1 方法一
线程：
from threading import Thread
from multiprocessing import Process
def work(n):
print('%s is running' %n)
if __name__ == '__main__':
t=Thread(target=work,args=(1,))
# t=Process(target=work,args=(1,))
t.start()
print('主线程')
进程：
1.3.2 方法二
from threading import Thread
from multiprocessing import Process
class MyThread(Thread):
def __init__(self,n):
super().__init__()
self.n=n
def run(self):
print('%s is running' % self.n)
if __name__ == '__main__':
t=MyThread(2)
t.start()
print('主线程')
2 进程和线程的区别
2.1 进程和线程的区别
1、线程共享创建它的进程的地址空间;进程有自己的地址空间。
2、线程可以直接访问进程的数据，子进程独立复制父进程的数据。
3、线程可以与自己进程中的其他线程交流，进程必须使用进程间的通信才能与其他进程交流。
4、新的线程很容易创建，新的进程需要复制父进程。
5、线程可以控制本进程中的其他线程，进程只能控制子进程
6、改变主线程可能影响进程中其他线程的行为，改变父进程不影响子进程
2.2 "子"线程与主线程的pid都是一样的
from threading import Thread
from multiprocessing import Process
import os
def work():
print('%s is running' %os.getpid())
if __name__ == '__main__':
t=Thread(target=work)
t=Process(target=work)
t.start()
print('主线程',os.getpid())
2.3 同一进程内的多个线程共享该进程的资源
from threading import Thread
from multiprocessing import Process
import os
n=100
def work():
global n
n=0
if __name__ == '__main__':
t=Thread(target=work)
# t=Process(target=work)
t.start()
t.join()
print('主线程',n)
3 为什么要用多线程
多线程指的是，在一个进程中开启多个线程，简单的讲：如果多个任务共用一块地址空间，那么必须在一个进程内开启多个线程
详细的说：
1、多线程共享一个进程的地址空间
2. 线程比进程更轻量级，线程比进程更容易创建可撤销，在许多操作系统中，创建一个线程比创建一个进程要快10-100倍，在有大量线程需要动态和快速修改时，这一特性很有用。
3. 若多个线程都是cpu密集型的，那么并不能获得性能上的增强，但是如果存在大量的计算和大量的I/O处理，拥有多个线程允许这些活动彼此重叠运行，从而会加快程序执行的速度。
4. 在多cpu系统中，为了最大限度的利用多核，可以开启多个线程，比开进程开销要小的多。（这一条并不适用于python）
4、守护线程
#主线程从执行角度就代表类该进程，主线程会在所有非守护线程都运行完毕才结束，守护线程就在主线程结束后结束
4.1 例一
(1)
from threading import Thread,current_thread
import time
def work():
print('%s is runing' %current_thread().getName())
time.sleep(2)
print('%s is done' %current_thread().getName())
if __name__ == '__main__':
t=Thread(target=work)
t.daemon=True
t.start()
print('主')
(2)
4.2 例二
(1)
from threading import Thread
import time
def foo():
print(123)
time.sleep(5)
print("end123")
def bar():
print(456)
time.sleep(3)
print("end456")
if __name__ == '__main__':
t1=Thread(target=foo)
t2=Thread(target=bar)
t1.daemon=True
t1.start()
t2.start()
print("main-------")
(2)
(3)
5、GIL
http://www.cnblogs.com/linhaifeng/articles/7449853.html#_label2
5.1 介绍
明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势。
GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。
解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据100，可能线程1执行x=100的同时，而垃圾回收执行的是回收100的操作，解决这种问题没有什么高明的方法，就是加锁处理，如下图的GIL，保证python解释器同一时间只能执行一个任务的代码
5.2 GIL与Lock
GIL保护的是解释器级的数据，保护用户自己的数据则需要自己加锁处理，如下图
5.3 多线程与多进程
#分析：
我们有四个任务需要处理，处理方式肯定是要玩出并发的效果，解决方案可以是：
方案一：开启四个进程
方案二：一个进程下，开启四个线程
#单核情况下，分析结果:
如果四个任务是计算密集型，没有多核来并行计算，方案一徒增了创建进程的开销，方案二胜
如果四个任务是I/O密集型，方案一创建进程的开销大，且进程的切换速度远不如线程，方案二胜
#多核情况下，分析结果：
如果四个任务是计算密集型，多核意味着并行计算，在python中一个进程中同一时刻只有一个线程执行用不上多核，方案一胜
如果四个任务是I/O密集型，再多的核也解决不了I/O问题，方案二胜
#结论：现在的计算机基本上都是多核，python对于计算密集型的任务开多线程的效率并不能带来多大性能上的提升，甚至不如串行(没有大量切换)，但是，对于IO密集型的任务效率还是有显著提升的。
应用：
多线程用于IO密集型，如socket，爬虫，web
多进程用于计算密集型，如金融分析
5.4 多线程与多进程性能测试
5.4.1 计算密集型
计算密集型：多进程效率高
(1)多线程
from multiprocessing import Process
from threading import Thread
import os,time
def work():
res=0
for i in range(10000000):
res*=i
if __name__ == '__main__':
l=[]
print(os.cpu_count()) #本机为4核
start=time.time()
for i in range(4):
# p=Process(target=work)
p=Thread(target=work)
l.append(p)
p.start()
for p in l:
p.join()
stop=time.time()
print('run time is %s' %(stop-start))
(2)多进程
from multiprocessing import Process
from threading import Thread
import os,time
def work():
res=0
for i in range(10000000):
res*=i
if __name__ == '__main__':
l=[]
print(os.cpu_count()) #本机为4核
start=time.time()
for i in range(4):
p=Process(target=work)
# p=Thread(target=work)
l.append(p)
p.start()
for p in l:
p.join()
stop=time.time()
print('run time is %s' %(stop-start))
5.4.2 I/O密集型
I/O密集型：多线程效率高
(1)多线程
from multiprocessing import Process
from threading import Thread
import threading
import os,time
def work():
time.sleep(2)
print('===>')
if __name__ == '__main__':
l=[]
print(os.cpu_count()) #本机为4核
start=time.time()
for i in range(400):
# p=Process(target=work)
p=Thread(target=work)
l.append(p)
p.start()
for p in l:
p.join()
stop=time.time()
print('run time is %s' %(stop-start))
(2)多进程
from multiprocessing import Process
from threading import Thread
import threading
import os,time
def work():
time.sleep(2)
print('===>')
if __name__ == '__main__':
l=[]
print(os.cpu_count()) #本机为4核
start=time.time()
for i in range(400):
p=Process(target=work)
# p=Thread(target=work)
l.append(p)
p.start()
for p in l:
p.join()
stop=time.time()
print('run time is %s' %(stop-start))
6、互斥锁
6.1 注意
#1.线程抢的是GIL锁，GIL锁相当于执行权限，拿到执行权限后才能拿到互斥锁Lock，其他线程也可以抢到GIL，但如果发现Lock仍然没有被释放则阻塞，即便是拿到执行权限GIL也要立刻交出来
#2.join是等待所有，即整体串行，而锁只是锁住修改共享数据的部分，即部分串行，要想保证数据安全的根本原理在于让并发变成串行，join与互斥锁都可以实现，毫无疑问，互斥锁的部分串行效率要更高
6.2 GIL VS Lock
6.2.1 GIL锁与互斥锁综合分析
首先我们需要达成共识：锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据
然后，我们可以得出结论：保护不同的数据就应该加不同的锁。
最后，问题就很明朗了，GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock
过程分析：所有线程抢的是GIL锁，或者说所有线程抢的是执行权限
线程1抢到GIL锁，拿到执行权限，开始执行，然后加了一把Lock，还没有执行完毕，即线程1还未释放Lock，有可能线程2抢到GIL锁，开始执行，执行过程中发现Lock还没有被线程1释放，于是线程2进入阻塞，被夺走执行权限，有可能线程1拿到GIL，然后正常执行到释放Lock。。。这就导致了串行运行的效果
6.2.2 互斥锁与join()的区别
既然是串行，那我们执行
t1.start()
t1.join
t2.start()
t2.join()
这也是串行执行啊，为何还要加Lock呢，需知join是等待t1所有的代码执行完，相当于锁住了t1的所有代码，而Lock只是锁住一部分操作共享数据的代码。
6.3 互斥锁用法
from threading import Thread,Lock
import time,random
n=100
def work():
global n
with mutex:
temp=n
time.sleep(0.1)
#如果在这睡了0.1秒，那么所有线程都拿到了n=100
#不加锁结果会变成99
n=temp-1
if __name__ == '__main__':
t_l=[]
mutex=Lock()
for i in range(100):
t=Thread(target=work)
t_l.append(t)
t.start()
for obj in t_l:
obj.join()
print(n)
7、死锁现象和递归锁
7.1 死锁
所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。
from threading import Thread,Lock,RLock
import time
mutexA=Lock()
mutexB=Lock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
mutexA.acquire()
print('%s 抢 A锁' %self.name)
mutexB.acquire()
print('%s 抢 B锁' %self.name)
mutexB.release()
mutexA.release()
def f2(self):
mutexB.acquire()
print('%s 抢 B锁' %self.name)
time.sleep(0.1)
mutexA.acquire()
print('%s 抢 A锁' %self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t=MyThread()
t.start()
第一个进程需要拿到A锁，但是A锁在第二个进程手里，第二个进程需要拿到B锁，但是B锁在第一个进程手里
7.2 递归锁
解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：
mutexA=mutexB=threading.RLock()
#一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止
from threading import Thread,Lock,RLock
import time
# mutexA=Lock()
# mutexB=Lock()
#递归锁的特点：
#1：可以acquire多次
#2：每acquire一次，计数加1，只要计数不为0，就不能被其他线程抢到
#互斥锁的特点：
#1：只能acquire一次
mutexA=mutexB=RLock()
# mutexA=mutexB=Lock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
mutexA.acquire()
print('%s 抢 A锁' %self.name)
mutexB.acquire()
print('%s 抢 B锁' %self.name)
mutexB.release()
mutexA.release()
def f2(self):
mutexB.acquire()
print('%s 抢 B锁' %self.name)
time.sleep(0.1)
mutexA.acquire()
print('%s 抢 A锁' %self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t=MyThread()
t.start()
8、concurrent.futures模块
https://docs.python.org/dev/library/concurrent.futures.html
8.1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
8.2 进程池与线程池
8.2.1 进程池
# 进程池
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,random,os
def work(n):
print('%s is running' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor()
futrues=[]
for i in range(10):
future=executor.submit(work,i)
# print(executor.submit(work,i).result()) #模拟同步的效果
futrues.append(future)
executor.shutdown(wait=True)
print('主')
for obj in futrues:
print(obj.result())
8.2.2 线程池
#线程池
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,random,os
import threading
def work(n):
print('%s is running' %threading.current_thread().getName())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ThreadPoolExecutor()
futrues=[]
for i in range(30):
future=executor.submit(work,i)
futrues.append(future)
executor.shutdown(wait=True)
print('主')
for obj in futrues:
print(obj.result())
8.3 用法
8.3.1 submit(fn,*args,**kwargs)
异步提交任务
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,random,os
def work(n):
print('%s is running' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor()
futrues=[]
for i in range(10):
future=executor.submit(work,i)
# print(executor.submit(work,i).result()) #模拟同步的效果
futrues.append(future)
executor.shutdown(wait=True)
print('主')
for obj in futrues:
print(obj.result())
8.3.2 shutdown(wait=True)
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前
#wait=True
#wait=True
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,random,os
def work(n):
print('%s is running' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor()
futrues=[]
for i in range(10):
future=executor.submit(work,i)
# print(executor.submit(work,i).result()) #模拟同步的效果
futrues.append(future)
executor.shutdown(wait=True)
print('主')
for obj in futrues:
print(obj.result())
#wait=False
#wait=False
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,random,os
def work(n):
print('%s is running' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor()
futrues=[]
for i in range(10):
future=executor.submit(work,i)
# print(executor.submit(work,i).result()) #模拟同步的效果
futrues.append(future)
executor.shutdown(wait=False)
print('主')
for obj in futrues:
print(obj.result())
8.3.3 result(timeout=None)
取得结果 想当与Pool中的get()
8.3.4 map(func, *iterables, timeout=None, chunksize=1)
取代for循环submit的操作
#map方法的使用
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,random,os
def work(n):
print('%s is running' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor()
# futures=[]
# for i in range(10):
# future=executor.submit(work,i)
# futures.append(future)
executor.map(work,range(10))
executor.shutdown(wait=True)
print('主')
8.3.5 add_done_callback(fn)
回调函数
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import requests
import os
import time,random
def get(url):
print('%s GET %s' %(os.getpid(),url))
response=requests.get(url)
time.sleep(random.randint(1,3))
if response.status_code == 200:
print('%s DONE %s' % (os.getpid(), url))
return {'url':url,'text':response.text}
def parse(future):
dic=future.result()
print('%s PARSE %s' %(os.getpid(),dic['url']))
time.sleep(1)
res='%s:%s
' %(dic['url'],len(dic['text']))
with open('db.txt','a') as f:
f.write(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=ProcessPoolExecutor()
start_time=time.time()
objs=[]
for url in urls:
# obj=p.apply_async(get,args=(url,),callback=parse) #主进程负责干回调函数的活
p.submit(get,url).add_done_callback(parse)
# p.close()
# p.join()
print('主',(time.time()-start_time))
9、信号量Semaphore
同进程的一样
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程
实例：(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)：
from threading import Semaphore,Thread,current_thread
import time,random
def task():
with sm:
print('%s 正在上厕所' %current_thread().getName())
time.sleep(random.randint(1,3))
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(11):
t=Thread(target=task)
t.start()
10、事件Event
10.1 Event
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
10.2 用法
10.2.1 event.isSet()
返回event的状态值；
10.2.2 event.wait()
如果 event.isSet()==False将阻塞线程；
10.2.3 event.set()
设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
10.2.4 event.clear()
恢复event的状态值为False。
10.3 实例
from threading import Event,current_thread,Thread
import time
e=Event()
def check():
print('%s 正在检测' %current_thread().getName())
time.sleep(5)
e.set()
def conn():
count=1
while not e.is_set():
if count > 3:
raise TimeoutError('连接超时')
print('%s 正在等待%s连接' % (current_thread().getName(),count))
e.wait(timeout=1)
count+=1
print('%s 开始连接' % current_thread().getName())
if __name__ == '__main__':
t1=Thread(target=check)
t2=Thread(target=conn)
t3=Thread(target=conn)
t4=Thread(target=conn)
t1.start()
t2.start()
t3.start()
t4.start()
11、定时器
定时器，指定n秒后执行某操作
12、线程队列queue
queue队列 ：使用import queue，用法与进程Queue一样
12.1 queue.Queue(maxsize=0)
#先进先出
12.2 queue.LifoQueue(maxsize=0)
#last in first后进先出，堆栈
12.3 queue.PriorityQueue(maxsize=0)
#存储数据时可设置优先级的队列
14、并发编程之协程
http://www.cnblogs.com/linhaifeng/articles/7429894.html
14.1 并发的本质
并发的本质是切换+保存状态， cpu正在运行一个任务，会在两种情况下切走去执行其他的任务（切换由操作系统强制控制），一种情况是该任务发生了阻塞，另外一种情况是该任务计算的时间过长。
对于单线程下，我们不可避免程序中出现io操作，但如果我们能在自己的程序中（即用户程序级别，而非操作系统级别）控制单线程下的多个任务能在一个任务遇到io阻塞时就切换到另外一个任务去计算，这样就保证了该线程能够最大限度地处于就绪态，即随时都可以被cpu执行的状态，相当于我们在用户程序级别将自己的io操作最大限度地隐藏起来，从而可以迷惑操作系统，让其看到：该线程好像是一直在计算，io比较少，从而更多的将cpu的执行权限分配给我们的线程。
协程的本质就是在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，以此来提升效率。
14.2 协程介绍
14.2.1 协程介绍
协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。
一句话说明什么是线程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。
#1. python的线程属于内核级别的，即由操作系统控制调度（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其他线程运行）
#2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，以此来提升效率（！！！非io操作的切换与效率无关）
14.2.2 协程优缺点
优点如下：
#1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
#2. 单线程内就可以实现并发的效果，最大限度地利用cpu
缺点如下：
#1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
#2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程
14.2.3 协程特点
1.必须在只有一个单线程里实现并发
2.修改共享数据不需加锁
3.用户程序里自己保存多个控制流的上下文栈
4.附加：一个协程遇到IO操作自动切换到其它协程（如何实现检测IO，yield、greenlet都无法实现，就用到了gevent模块（select机制））
14.3 greenlet模块
如果我们在单个线程内有20个任务，要想实现在多个任务之间切换，使用yield生成器的方式过于麻烦（需要先得到初始化一次的生成器，然后再调用send。。。非常麻烦），而使用greenlet模块可以非常简单地实现这20个任务直接的切换
from greenlet import greenlet
def eat(name):
print('%s eat 1' %name)
g2.switch('egon')
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('egon')#可以在第一次switch时传入参数，以后都不需要
greenlet只是提供了一种比generator更加便捷的切换方式，当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。
14.4 gevent模块
Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。
14.4.1 gevent.spawn(func,1,,2,3,x=4,y=5)
g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的
14.4.2 gevent.spawn(func2)
g2=gevent.spawn(func2)
14.4.3 gevent.joinall([g1,g2])
g1.join() #等待g1结束
g2.join() #等待g2结束
#或者上述两步合作一步：gevent.joinall([g1,g2])
14.4.4 g1.value
g1.value#拿到func1的返回值
14.4.5 遇到IO阻塞时会自动切换任务
import gevent
def eat(name):
print('%s eat 1' %name)
gevent.sleep(2)
print('%s eat 2' %name)
def play(name):
print('%s play 1' %name)
gevent.sleep(1)
print('%s play 2' %name)
g1=gevent.spawn(eat,'egon')
g2=gevent.spawn(play,name='egon')
g1.join()
g2.join()
#或者gevent.joinall([g1,g2])
print('主')
上例gevent.sleep(2)模拟的是gevent可以识别的io阻塞,
而time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了
from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面，如time，socket模块之前
或者我们干脆记忆成：要用gevent，需要将from gevent import monkey;monkey.patch_all()放到文件的开头
from gevent import monkey;monkey.patch_all()
import gevent
import time
def eat():
print('eat food 1')
time.sleep(2)
print('eat food 2')
def play():
print('play 1')
time.sleep(1)
print('play 2')
g1=gevent.spawn(eat)
g2=gevent.spawn(play)
gevent.joinall([g1,g2])
print('主')
我们可以用threading.current_thread().getName()来查看每个g1和g2，查看的结果为DummyThread-n，即假线程
14.4.6 Gevent之应用举例一
爬虫
from gevent import monkey;monkey.patch_all()
import gevent
import requests
from threading import current_thread
def get(url):
print('%s get %s' %(current_thread().getName(),url))
response=requests.get(url)
if response.status_code == 200:
return {'url':len(response.text)}
# print({'url':len(response.text)})
g1=gevent.spawn(get,'http://www.baidu.com')
g2=gevent.spawn(get,'http://www.python.org')
g3=gevent.spawn(get,'http://www.jd.com')
g1.join()
g2.join()
g3.join()
print(g1.value)
print(g2.value)
print(g3.value)
14.4.7 Gevent之应用举例二
通过gevent实现单线程下的socket并发（from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前，否则gevent无法识别socket的阻塞）
服务端：
from gevent import monkey;monkey.patch_all()
import gevent
from multiprocessing import Process
from socket import *
def server(ip,port):
s = socket(AF_INET, SOCK_STREAM)
s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
s.bind((ip,port))
s.listen(5)
while True:
conn,addr=s.accept()
print('%s:%s' % (addr[0], addr[1]))
g1=gevent.spawn(talk,conn,addr)
def talk(conn,addr):
while True:
try:
data=conn.recv(1024)
print('%s:%s [%s]' %(addr[0],addr[1],data))
if not data:break
conn.send(data.upper())
except ConnectionResetError:
break
conn.close()
if __name__ == '__main__':
server('127.0.0.1',8090)
客户端：
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8090))
while True:
msg=input('>>: ').strip()
if not msg:continue
c.send(msg.encode('utf-8'))
data=c.recv(1024)
print(data.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()