python并发编程补充

一、进程通信

1、信号量

互斥锁：同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据

如果指定信号量为3，那么来一个人获得一把锁，计数加1，当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁，信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念。

<code>from</code> <code>multiprocessing </code><code>import</code> <code>Process,Semaphore</code>

<code>import</code> <code>time,random</code>

<code>def</code> <code>action(sem,user):</code>

<code> </code><code>sem.acquire()</code>

<code> </code><code>print</code><code>(</code><code>'%s 占一个位置'</code> <code>%</code><code>user)</code>

<code> </code><code>time.sleep(random.randint(</code><code>0</code><code>,</code><code>3</code><code>)) </code><code>#模拟进程执行时间</code>

<code> </code><code>sem.release()</code>

<code> </code><code>sem</code><code>=</code><code>Semaphore(</code><code>5</code><code>)</code>

<code> </code><code>p_l</code><code>=</code><code>[] </code><code>#存放启动的进程</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(</code><code>13</code><code>):</code>

<code> </code><code>p</code><code>=</code><code>Process(target</code><code>=</code><code>action,args</code><code>=</code><code>(sem,</code><code>'user%s'</code> <code>%</code><code>i,))</code>

<code> </code><code>p.start()</code>

<code> </code><code>p_l.append(p)</code>

<code> </code><code>i.join() </code><code>#保证所以子进程执行完毕</code>

<code> </code><code>print</code><code>(</code><code>'============》'</code><code>)</code>

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个客户端处于等待状态

# 服务端

<code>from</code> <code>socket </code><code>import</code> <code>*</code>

<code>from</code> <code>multiprocessing </code><code>import</code> <code>Pool</code>

<code>import</code> <code>os</code>

<code>server</code><code>=</code><code>socket(AF_INET,SOCK_STREAM)</code>

<code>server.setsockopt(SOL_SOCKET,SO_REUSEADDR,</code><code>1</code><code>)</code>

<code>server.bind((</code><code>'127.0.0.1'</code><code>,</code><code>8080</code><code>))</code>

<code>server.listen(</code><code>5</code><code>)</code>

<code>def</code> <code>talk(conn,client_addr):</code>

<code> </code><code>print</code><code>(</code><code>'进程pid: %s'</code> <code>%</code><code>os.getpid())</code>

<code> </code><code>while</code> <code>True</code><code>:</code>

<code> </code><code>if</code> <code>not</code> <code>msg:</code><code>break</code>

<code> </code><code>conn.send(msg.upper())</code>

<code> </code><code>except</code> <code>Exception:</code>

<code> </code><code>break</code>

<code> </code><code>p</code><code>=</code><code>Pool() </code><code>#Pool内的进程数默认是cpu核数，假设为4（查看方法os.cpu_count()）</code>

<code> </code><code>conn,client_addr</code><code>=</code><code>server.accept()</code>

<code> </code><code>p.apply_async(talk,args</code><code>=</code><code>(conn,client_addr))</code>

<code> </code><code># p.apply(talk,args=(conn,client_addr)) #同步的话，则同一时间只有一个客户端能访问</code>

# 客户端

<code>client</code><code>=</code><code>socket(AF_INET,SOCK_STREAM)</code>

<code>client.connect((</code><code>'127.0.0.1'</code><code>,</code><code>8080</code><code>))</code>

<code>while</code> <code>True</code><code>:</code>

<code> </code><code>msg</code><code>=</code><code>input</code><code>(</code><code>'>>: '</code><code>).strip()</code>

<code> </code><code>if</code> <code>not</code> <code>msg:</code><code>continue</code>

<code> </code><code>client.send(msg.encode(</code><code>'utf-8'</code><code>))</code>

<code> </code><code>msg</code><code>=</code><code>client.recv(</code><code>1024</code><code>)</code>

<code> </code><code>print</code><code>(msg.decode(</code><code>'utf-8'</code><code>))</code>

5、回调函数

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。

<code>import</code> <code>requests</code>

<code>import</code> <code>re</code>

<code>def</code> <code>get_page(url,pattern):</code>

<code> </code><code>response</code><code>=</code><code>requests.get(url)</code>

<code> </code><code>if</code> <code>response.status_code </code><code>=</code><code>=</code> <code>200</code><code>:</code>

<code> </code><code>return</code> <code>(response.text,pattern)</code>

<code>def</code> <code>parse_page(info):</code>

<code> </code><code>page_content,pattern</code><code>=</code><code>info</code>

<code> </code><code>res</code><code>=</code><code>re.findall(pattern,page_content)</code>

<code> </code><code>'index'</code><code>:item[</code><code>0</code><code>],</code>

<code> </code><code>'title'</code><code>:item[</code><code>1</code><code>],</code>

<code> </code><code>'actor'</code><code>:item[</code><code>2</code><code>].strip()[</code><code>3</code><code>:],</code>

<code> </code><code>'score'</code><code>:item[</code><code>4</code><code>]</code><code>+</code><code>item[</code><code>5</code><code>]</code>

<code> </code><code>print</code><code>(dic)</code>

<code> </code><code>pattern1</code><code>=</code><code>re.</code><code>compile</code><code>(r</code><code>'<dd>.*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<'</code><code>,re.S)</code>

<code> </code><code>'http://maoyan.com/board/7'</code><code>:pattern1,</code>

<code> </code><code>for</code> <code>url,pattern </code><code>in</code> <code>url_dic.items():</code>

<code> </code><code>res</code><code>=</code><code>p.apply_async(get_page,args</code><code>=</code><code>(url,pattern),callback</code><code>=</code><code>parse_page)</code>

<code> </code><code>res_l.append(res)</code>

<code> </code><code># res=requests.get('http://maoyan.com/board/7')</code>

<code> </code><code># print(re.findall(pattern,res.text))</code>

<code>{'index': '1', 'title': '神秘巨星', 'actor': '阿米尔·汗,塞伊拉·沃西,梅·维贾', 'time': '2018-01-19', 'score': '9.5'}</code>

<code>{'index': '2', 'title': '奇迹男孩', 'actor': '雅各布·特瑞布雷,朱莉娅·罗伯茨,欧文·威尔逊', 'time': '2018-01-19', 'score': '9.3'}</code>

<code>{'index': '3', 'title': '小狗奶瓶', 'actor': '奶瓶,康潇诺,魏子涵', 'time': '2018-02-02', 'score': '9.3'}</code>

<code>{'index': '4', 'title': '公牛历险记', 'actor': '约翰·塞纳,莉莉·戴,凯特·迈克金农', 'time': '2018-01-19', 'score': '9.2'}</code>

<code>{'index': '5', 'title': '前任3：再见前任', 'actor': '韩庚,郑恺,于文文', 'time': '2017-12-29', 'score': '9.2'}</code>

<code>{'index': '6', 'title': '一个人的课堂', 'actor': '孙海英,韩三明,王乃训', 'time': '2018-01-16', 'score': '9.2'}</code>

<code>{'index': '7', 'title': '芳华', 'actor': '黄轩,苗苗,钟楚曦', 'time': '2017-12-15', 'score': '9.1'}</code>

<code>{'index': '8', 'title': '南极之恋', 'actor': '赵又廷,杨子姗', 'time': '2018-02-01', 'score': '9.0'}</code>

<code>{'index': '9', 'title': '马戏之王', 'actor': '休·杰克曼,扎克·埃夫隆,米歇尔·威廉姆斯', 'time': '2018-02-01', 'score': '9.0'}</code>

<code>{'index': '10', 'title': '小马宝莉大电影', 'actor': '奥卓·阿杜巴,艾米莉·布朗特,克里斯汀·肯诺恩斯', 'time': '2018-02-02', 'score': '8.9'}</code>

6、如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数

<code>import</code> <code>time,random,os</code>

<code> </code><code>time.sleep(</code><code>1</code><code>)</code>

<code> </code><code>return</code> <code>n</code><code>*</code><code>*</code><code>2</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(</code><code>10</code><code>):</code>

<code> </code><code>res</code><code>=</code><code>p.apply_async(work,args</code><code>=</code><code>(i,))</code>

<code> </code><code>p.close()</code>

<code> </code><code>p.join() </code><code>#等待进程池中所有进程执行完毕</code>

<code> </code><code>nums.append(res.get()) </code><code>#拿到所有结果</code>

<code> </code><code>print</code><code>(nums) </code><code>#主进程拿到所有的处理结果,可以在主进程中进行统一进行处理</code>

二、通信线程

1、死锁现象与递归锁

所谓死锁：是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程

解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。

<code>from</code> <code>threading </code><code>import</code> <code>Thread,RLock</code>

<code>import</code> <code>time</code>

<code>mutexA</code><code>=</code><code>mutexB</code><code>=</code><code>RLock()</code>

<code>class</code> <code>MyThread(Thread):</code>

<code> </code><code>mutexA.acquire()</code>

<code> </code><code>print</code><code>(</code><code>'\033[41m%s 拿到A锁\033[0m'</code> <code>%</code><code>self</code><code>.name)</code>

<code> </code><code>mutexB.acquire()</code>

<code> </code><code>print</code><code>(</code><code>'\033[42m%s 拿到B锁\033[0m'</code> <code>%</code><code>self</code><code>.name)</code>

<code> </code><code>mutexB.release()</code>

<code> </code><code>mutexA.release()</code>

<code> </code><code>print</code><code>(</code><code>'\033[43m%s 拿到B锁\033[0m'</code> <code>%</code><code>self</code><code>.name)</code>

<code> </code><code>time.sleep(</code><code>2</code><code>)</code>

<code> </code><code>print</code><code>(</code><code>'\033[44m%s 拿到A锁\033[0m'</code> <code>%</code><code>self</code><code>.name)</code>

<code> </code><code>t</code><code>=</code><code>MyThread()</code>

<code> </code><code>t.start()</code>

2、信号量Semaphore

同进程的一样

Semaphore管理一个内置的计数器，每当调用acquire()时内置计数器-1；调用release() 时内置计数器+1；

计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

<code>from</code> <code>threading </code><code>import</code> <code>Thread,Semaphore,current_thread</code>

<code>import</code> <code>threading</code>

<code> </code><code>print</code><code>(</code><code>'%s get sm'</code> <code>%</code><code>current_thread().getName())</code>

<code> </code><code>time.sleep(random.randint(</code><code>1</code><code>,</code><code>3</code><code>))</code>

<code> </code><code>sm</code><code>=</code><code>Semaphore(</code><code>5</code><code>)</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(</code><code>20</code><code>):</code>

<code> </code><code>t</code><code>=</code><code>Thread(target</code><code>=</code><code>func)</code>

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。

<code># 模拟连接mysql</code>

<code>from</code> <code>threading </code><code>import</code> <code>Thread,Event</code>

<code>def</code> <code>conn_mysql():</code>

<code> </code><code>count</code><code>=</code><code>1</code>

<code> </code><code>while</code> <code>not</code> <code>event.is_set():</code>

<code> </code><code>if</code> <code>count > </code><code>3</code><code>: </code><code>#超过次数就抛出异常链接超时</code>

<code> </code><code>raise</code> <code>TimeoutError(</code><code>'链接超时'</code><code>)</code>

<code> </code><code>print</code><code>(</code><code>'<%s>第%s次尝试链接'</code> <code>%</code> <code>(threading.current_thread().getName(), count))</code>

<code> </code><code>event.wait(</code><code>1</code><code>) </code><code>#等待1秒后接着尝试连接</code>

<code> </code><code>count</code><code>+</code><code>=</code><code>1</code>

<code> </code><code>print</code><code>(</code><code>'<%s>链接成功'</code> <code>%</code><code>threading.current_thread().getName())</code>

<code>def</code> <code>check_mysql():</code>

<code> </code><code>print</code><code>(</code><code>'\033[45m[%s]正在检查mysql\033[0m'</code> <code>%</code> <code>threading.current_thread().getName())</code>

<code> </code><code>time.sleep(random.randint(</code><code>2</code><code>,</code><code>4</code><code>))</code>

<code> </code><code>event.</code><code>set</code><code>()</code>

<code> </code><code>event</code><code>=</code><code>Event()</code>

<code> </code><code>conn1</code><code>=</code><code>Thread(target</code><code>=</code><code>conn_mysql)</code>

<code> </code><code>conn2</code><code>=</code><code>Thread(target</code><code>=</code><code>conn_mysql)</code>

<code> </code><code>check</code><code>=</code><code>Thread(target</code><code>=</code><code>check_mysql)</code>

<code> </code><code>conn1.start()</code>

<code> </code><code>conn2.start()</code>

<code> </code><code>check.start()</code>

4、定时器

# 验证码定时器

<code>from</code> <code>threading </code><code>import</code> <code>Timer</code>

<code>import</code> <code>random,time</code>

<code>class</code> <code>Code:</code>

<code> </code><code>self</code><code>.make_cache()</code>

<code> </code><code>def</code> <code>make_cache(</code><code>self</code><code>,interval</code><code>=</code><code>5</code><code>):</code>

<code> </code><code>self</code><code>.cache</code><code>=</code><code>self</code><code>.make_code()</code>

<code> </code><code>print</code><code>(</code><code>self</code><code>.cache)</code>

<code> </code><code>self</code><code>.t</code><code>=</code><code>Timer(interval,</code><code>self</code><code>.make_cache)</code>

<code> </code><code>self</code><code>.t.start()</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(n):</code>

<code> </code><code>s1</code><code>=</code><code>str</code><code>(random.randint(</code><code>0</code><code>,</code><code>9</code><code>))</code>

<code> </code><code>s2</code><code>=</code><code>chr</code><code>(random.randint(</code><code>65</code><code>,</code><code>90</code><code>))</code>

<code> </code><code>res</code><code>+</code><code>=</code><code>random.choice([s1,s2])</code>

<code> </code><code>return</code> <code>res</code>

<code> </code><code>def</code> <code>check(</code><code>self</code><code>):</code>

<code> </code><code>while</code> <code>True</code><code>:</code>

<code> </code><code>inp</code><code>=</code><code>input</code><code>(</code><code>'>>: '</code><code>).strip()</code>

<code> </code><code>if</code> <code>inp.upper() </code><code>=</code><code>=</code> <code>self</code><code>.cache:</code>

<code> </code><code>print</code><code>(</code><code>'验证成功'</code><code>,end</code><code>=</code><code>'\n'</code><code>)</code>

<code> </code><code>self</code><code>.t.cancel()</code>

<code> </code><code>break</code>

<code> </code><code>obj.check()</code>

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）堆栈（后进先出）

q=queue.LifoQueue()

（3）按照优先级取值

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用法

<code>from</code> <code>concurrent.futures </code><code>import</code> <code>ThreadPoolExecutor,ProcessPoolExecutor</code>

<code>import</code> <code>os,time,random</code>

<code> </code><code>print</code><code>(</code><code>'%s is runing'</code> <code>%</code><code>os.getpid())</code>

<code> </code><code>time.sleep(random.randint(</code><code>1</code><code>,</code><code>3</code><code>))</code>

<code> </code><code>executor</code><code>=</code><code>ProcessPoolExecutor(max_workers</code><code>=</code><code>3</code><code>)</code>

<code> </code><code>futures</code><code>=</code><code>[]</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(</code><code>11</code><code>):</code>

<code> </code><code>future</code><code>=</code><code>executor.submit(task,i) </code><code>#异步提交任务</code>

<code> </code><code>futures.append(future)</code>

<code> </code><code>executor.shutdown(</code><code>True</code><code>) </code><code>#wait=True，等待池内所有任务执行完毕回收完资源后才继续，wait=False，立即返回，并不会等待池内的任务执行完毕</code>

<code> </code><code>print</code><code>(</code><code>'+++>'</code><code>)</code>

<code> </code><code>for</code> <code>future </code><code>in</code> <code>futures:</code>

<code> </code><code>print</code><code>(future.result()) </code><code>#取得结果</code>

3、map的用法

<code> </code><code>executor</code><code>=</code><code>ThreadPoolExecutor(max_workers</code><code>=</code><code>3</code><code>)</code>

<code> </code><code># for i in range(11):</code>

<code> </code><code># future=executor.submit(task,i)</code>

<code> </code><code>executor.</code><code>map</code><code>(task,</code><code>range</code><code>(</code><code>1</code><code>,</code><code>10</code><code>)) </code><code>#map取代for循环submit的操作</code>

4、回调函数

<code>import</code> <code>json</code>

<code> </code><code>print</code><code>(</code><code>'<进程%s> get %s'</code> <code>%</code><code>(os.getpid(),url))</code>

<code> </code><code>respone</code><code>=</code><code>requests.get(url)</code>

<code> </code><code>if</code> <code>respone.status_code </code><code>=</code><code>=</code> <code>200</code><code>:</code>

<code> </code><code>return</code> <code>{</code><code>'url'</code><code>:url,</code><code>'text'</code><code>:respone.text}</code>

<code>def</code> <code>parse_page(res):</code>

<code> </code><code>res</code><code>=</code><code>res.result()</code>

<code> </code><code>print</code><code>(</code><code>'<进程%s> parse %s'</code> <code>%</code><code>(os.getpid(),res[</code><code>'url'</code><code>]))</code>

<code> </code><code>parse_res</code><code>=</code><code>'url:<%s> size:[%s]\n'</code> <code>%</code><code>(res[</code><code>'url'</code><code>],</code><code>len</code><code>(res[</code><code>'text'</code><code>]))</code>

<code> </code><code>f.write(parse_res)</code>

<code> </code><code>'https://www.baidu.com'</code><code>,</code>

<code> </code><code>'https://www.python.org'</code><code>,</code>

<code> </code><code>'https://www.openstack.org'</code><code>,</code>

<code> </code><code>'https://help.github.com/'</code><code>,</code>

<code> </code><code># p.apply_async(get_page,args=(url,),callback=pasrse_page)</code>

<code> </code><code># p.close()</code>

<code> </code><code>p</code><code>=</code><code>ProcessPoolExecutor(</code><code>3</code><code>)</code>

<code> </code><code>p.submit(get_page,url).add_done_callback(parse_page) </code><code>#parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果</code>

5、同步调用和异步调用

提交任务的两种方式：

同步调用：提交完任务后，就在原地等待，等待任务执行完毕，拿到任务的返回值，才能继续下一行代码，导致程序串行执行

异步调用+回调机制：提交完任务后，不在原地等待，任务一旦执行完毕就会触发回调函数的执行，程序是并发执行

#同步调用示例：

<code> </code><code>print</code><code>(</code><code>'%s is ruuning'</code> <code>%</code><code>os.getpid())</code>

<code>def</code> <code>handle(res):</code>

<code> </code><code>print</code><code>(</code><code>'handle res %s'</code> <code>%</code><code>res)</code>

<code> </code><code>pool</code><code>=</code><code>ProcessPoolExecutor(</code><code>2</code><code>)</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(</code><code>5</code><code>):</code>

<code> </code><code>res</code><code>=</code><code>pool.submit(task,i).result()</code>

<code> </code><code>handle(res)</code>

<code> </code><code>pool.shutdown(wait</code><code>=</code><code>True</code><code>)</code>

<code> </code><code># pool.submit(task,33333)</code>

<code> </code><code>print</code><code>(</code><code>'主'</code><code>)</code>

# 异步调用示例：

<code> </code><code># handle(res)</code>

<code> </code><code>obj</code><code>=</code><code>pool.submit(task,i)</code>

<code> </code><code>obj.add_done_callback(handle)</code>

四、协程

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模块

<code>#pip3 install greenlet #安装greenlet模块</code>

<code>from</code> <code>greenlet </code><code>import</code> <code>greenlet</code>

<code> </code><code>print</code><code>(</code><code>'%s eat 1'</code> <code>%</code><code>name)</code>

<code> </code><code>time.sleep(</code><code>100</code><code>)</code>

<code> </code><code>g2.switch(</code><code>'wang'</code><code>)</code>

<code> </code><code>print</code><code>(</code><code>'%s eat 2'</code> <code>%</code><code>name)</code>

<code> </code><code>g2.switch()</code>

<code> </code><code>print</code><code>(</code><code>'%s play 1'</code> <code>%</code> <code>name)</code>

<code> </code><code>g1.switch()</code>

<code> </code><code>print</code><code>(</code><code>'%s play 2'</code> <code>%</code> <code>name)</code>

<code>g1</code><code>=</code><code>greenlet(eat)</code>

<code>g2</code><code>=</code><code>greenlet(play)</code>

<code>g1.switch(</code><code>'wang'</code><code>)</code>

<code>使用greenlet模块可以非常简单地实现多个任务的切换，当切到一个任务执行时如果遇到io，那就原地阻塞，这仍然没有解决遇到IO自动切换来提升效率的问题</code>

5、Gevent模块(遇到IO阻塞时会自动切换任务)

#pip3 install gevent #安装Gevent模块

（1）用法

<code>g1</code><code>=</code><code>gevent.spawn(func,</code><code>1</code><code>,,</code><code>2</code><code>,</code><code>3</code><code>,x</code><code>=</code><code>4</code><code>,y</code><code>=</code><code>5</code><code>) </code><code>#创建一个协程对象g1，spawn括号内第一个参数是函数名，后面可以有多个位置实参或关键字实参，都是传给函数的</code>

<code>g2</code><code>=</code><code>gevent.spawn(func2)</code>

<code>#或者上述两步合作一步：gevent.joinall([g1,g2])</code>

<code>g1.value </code><code>#拿到func1的返回值</code>

（2）遇到IO阻塞时会自动切换任务

<code>from</code> <code>gevent </code><code>import</code> <code>monkey;monkey.patch_all()</code>

<code>import</code> <code>gevent</code>

<code> </code><code># gevent.sleep(3)</code>

<code> </code><code>time.sleep(</code><code>3</code><code>)</code>

<code> </code><code># gevent.sleep(2)</code>

<code>g1</code><code>=</code><code>gevent.spawn(eat,</code><code>'wang'</code><code>)</code>

<code>g2</code><code>=</code><code>gevent.spawn(play,</code><code>'li'</code><code>)</code>

<code># gevent.sleep(1)</code>

<code>gevent.joinall([g1,g2])</code>

<code>我们可以用threading.current_thread().getName()来查看每个g1和g2，查看的结果为DummyThread</code><code>-</code><code>n，即假线程</code>

（3）爬虫

<code> </code><code>print</code><code>(</code><code>'GET: %s'</code> <code>%</code><code>url)</code>

<code> </code><code>print</code><code>(</code><code>'%d bytes received from %s'</code> <code>%</code><code>(</code><code>len</code><code>(response.text),url)) </code><code>#统计内容的长度</code>

<code>start_time</code><code>=</code><code>time.time()</code>

<code>gevent.joinall([</code>

<code> </code><code>gevent.spawn(get_page,</code><code>'https://www.python.org/'</code><code>),</code>

<code> </code><code>gevent.spawn(get_page,</code><code>'https://www.yahoo.com/'</code><code>),</code>

<code> </code><code>gevent.spawn(get_page,</code><code>'https://github.com/'</code><code>),</code>

<code>print</code><code>(</code><code>'run time is %s'</code> <code>%</code><code>(stop_time</code><code>-</code><code>start_time))</code>

（4）通过gevent实现单线程下的socket并发（from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前，否则gevent无法识别socket的阻塞）

<code>#如果不想用money.patch_all()打补丁,可以用gevent自带的socket</code>

<code># from gevent import socket</code>

<code># s=socket.socket()</code>

<code>def</code> <code>server(server_ip,port):</code>

<code> </code><code>s</code><code>=</code><code>socket(AF_INET,SOCK_STREAM)</code>

<code> </code><code>s.setsockopt(SOL_SOCKET,SO_REUSEADDR,</code><code>1</code><code>)</code>

<code> </code><code>s.bind((server_ip,port))</code>

<code> </code><code>s.listen(</code><code>5</code><code>)</code>

<code> </code><code>conn,addr</code><code>=</code><code>s.accept()</code>

<code> </code><code>gevent.spawn(talk,conn,addr)</code>

<code> </code><code>print</code><code>(</code><code>'client %s:%s msg: %s'</code> <code>%</code><code>(addr[</code><code>0</code><code>],addr[</code><code>1</code><code>],res))</code>

<code> </code><code>conn.send(res.upper())</code>

<code> </code><code>except</code> <code>Exception as e:</code>

<code> </code><code>print</code><code>(e)</code>

<code> </code><code>finally</code><code>:</code>

<code> </code><code>conn.close()</code>

<code> </code><code>server(</code><code>'127.0.0.1'</code><code>,</code><code>8080</code><code>)</code>

<code>#_*_coding:utf-8_*_</code>

# 多线程并发多个客户端

<code>from</code> <code>threading </code><code>import</code> <code>Thread</code>

<code>def</code> <code>client(server_ip,port):</code>

<code> </code><code>c</code><code>=</code><code>socket(AF_INET,SOCK_STREAM) </code><code>#套接字对象一定要加到函数内，即局部名称空间内，放在函数外则被所有线程共享，那么客户端端口永远一样了</code>

<code> </code><code>c.connect((server_ip,port))</code>

<code> </code><code>count</code><code>=</code><code>0</code>

<code> </code><code>c.send((</code><code>'%s say hello %s'</code> <code>%</code><code>(threading.current_thread().getName(),count)).encode(</code><code>'utf-8'</code><code>))</code>

<code> </code><code>print</code><code>(msg.decode(</code><code>'utf-8'</code><code>))</code>

<code> </code><code>for</code> <code>i </code><code>in</code> <code>range</code><code>(</code><code>500</code><code>):</code>

<code> </code><code>t</code><code>=</code><code>Thread(target</code><code>=</code><code>client,args</code><code>=</code><code>(</code><code>'127.0.0.1'</code><code>,</code><code>8080</code><code>))</code>

本文转自宋鹏超 51CTO博客，原文链接：http://blog.51cto.com/qidian510/2071128，如需转载请自行联系原作者

python并发编程补充

继续阅读

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

Cloud Studio初体验

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

使用 ctypes 进行 Python 和 C 的混合编程

【python】【数据处理】画多维数据分布图

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【python】netconf协议对接管理设备

「Python 网络自动化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 网络设备

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method

在python中创建excel并写入