python 之 併發編程（線程Event、協程）

9.14 線程Event

connect線程執行到event.wait()時開始等待，直到check線程執行event.set()後立即繼續線程connect

from threading import Event,current_thread,Thread
import time
event=Event()
def check():
    print('%s 正在檢測服務是否正常....' %current_thread().name)
    time.sleep(3)
    event.set()
​
def connect():
    print('%s 等待連接...' %current_thread().name)
    event.wait()
    print('%s 開始連接...' % current_thread().name)
​
if __name__ == '__main__':
    t1=Thread(target=connect)
    t2=Thread(target=connect)
    t3=Thread(target=connect)
​
    c1=Thread(target=check)
​
    t1.start()
    t2.start()
    t3.start()
    c1.start()

connect線程執行到event.wait(1)時開始等待1秒，count計數+1，如果到check線程執行event.set()前已經4秒，則終止線程connect，否則event.is_set() is True ,立即繼續線程connect

from threading import Event,current_thread,Thread
import time
event=Event()
def check():
    print('%s 正在檢測服務是否正常....' %current_thread().name)
    time.sleep(5)
    event.set()
​
def connect():
    count=1
    while not event.is_set():   #event是否被set過，是返回True,否返回False
        if count ==  4:
            print('嘗試的次數過多，請稍後重試')
            return
        print('%s 嘗試第%s次連接...' %(current_thread().name,count))
        event.wait(1)
        count+=1
    print('%s 開始連接...' % current_thread().name)
​
if __name__ == '__main__':
    t1=Thread(target=connect)
    t2=Thread(target=connect)
    t3=Thread(target=connect)
​
    c1=Thread(target=check)
​
    t1.start()
    t2.start()
    t3.start()
    c1.start()

9.15 協程

協程：是單線程下的併發，又稱微線程，纖程。一句話說明什麼是線程：協程是一種用戶態的輕量級線程，即協程是由用戶程式自己控制調度的。

1. python的線程屬於內核級別的，即由操作系統控制調度（如單線程遇到i/o或執行時間過長就會被迫交出cpu執行許可權，切換其他線程運行）

2. 單線程內開啟協程，一旦遇到i/o，就會從應用程式級別（而非操作系統）控制切換到其他任務，以此來提升效率（非i/o操作的切換與效率無關）

3. 對比操作系統控制線程的切換，用戶在單線程內控制協程的切換

  優點:
  1. 協程的切換開銷更小，屬於程式級別的切換，操作系統完全感知不到，因而更加輕量級
  2. 單線程內就可以實現併發的效果，最大限度地利用cpu
     
  缺點：
  1. 協程的本質是單線程下，無法利用多核，可以是一個程式開啟多個進程，每個進程內開啟多個線程，每個線程內開啟協程
  2. 協程指的是單個線程，因而一旦協程出現阻塞，將會阻塞整個線程

9.151 greenlet模塊

 from greenlet import greenlet
  import time
  
  def eat(name):
      print('%s eat 1' %name)
      #time.sleep(30)           遇到i/o不能自動切換
      g2.switch('alex')
      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') # egon eat 1   alex play 1   egon eat 2   alex play 2

9.152 gevent模塊

import gevent
​
def eat(name):
    print('%s eat 1' %name)
    gevent.sleep(5)         #只檢測gevent的i/o
    print('%s eat 2' %name)
def play(name):
    print('%s play 1' %name)
    gevent.sleep(3)
    print('%s play 2' %name)
​
g1=gevent.spawn(eat,'egon') #非同步提交任務
g2=gevent.spawn(play,'alex')
​
# gevent.sleep(100)
# g1.join()
# g2.join()                # joinall等待任務執行完畢再結束線程
gevent.joinall([g1,g2])     # egon eat 1     alex play 1    alex play 2   egon eat 2

from gevent import monkey;monkey.patch_all()#標記所有(包括time等)的i/o
import gevent
import time
​
def eat(name):
    print('%s eat 1' %name)
    time.sleep(5)            #time的i/o也可以檢測
    print('%s eat 2' %name)
def play(name):
    print('%s play 1' %name)
    time.sleep(3)
    print('%s play 2' %name)
​
g1=gevent.spawn(eat,'egon')
g2=gevent.spawn(play,'alex')
​
# gevent.sleep(100)
# g1.join()
# g2.join()
gevent.joinall([g1,g2])       # egon eat 1   alex play 1    alex play 2   egon eat 2

驗證協程的假名：

from gevent import monkey;monkey.patch_all()#標記所有time等的i/o
from threading import current_thread
import gevent
import time
​
def eat():
    print('%s eat 1' %current_thread().name)
    time.sleep(5)
    print('%s eat 2' %current_thread().name)
def play():
    print('%s play 1' %current_thread().name)
    time.sleep(3)
    print('%s play 2' %current_thread().name)
​
g1=gevent.spawn(eat)
g2=gevent.spawn(play)
​
# gevent.sleep(100)
# g1.join()
# g2.join()
print(current_thread().name)#MainThread
gevent.joinall([g1,g2])     #DummyThread-1 eat 1     DummyThread-2 play 1  DummyThread-2 play 2                             DummyThread-1 eat 2

9.153 基於協程實現併發的套接字通信

服務端：

from gevent import monkey,spawn;monkey.patch_all()#標記所有time等的i/o
from threading import Thread
from socket import *
​
def talk(conn):
    while True:
        try:
            data=conn.recv(1024)
            if not data:break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()
​
def server(ip,port,backlog=5):
    s = socket()
    s.bind((ip,port))
    s.listen(backlog)
​
    while True:
        conn, addr = s.accept()
        print(addr)
      
        g=spawn(talk,conn)  # 通信
    s.close()
​
if __name__ == '__main__':
    spawn(server,'127.0.0.1',8080).join()
    # server(('127.0.0.1',8080))

客戶端：

from threading import Thread,current_thread
from socket import *
import os
​
def client():
    client = socket()
    client.connect(('127.0.0.1', 8080))
​
    while True:
        data = '%s hello' % current_thread().name
        client.send(data.encode('utf-8'))
        res = client.recv(1024)
        print(res.decode('utf-8'))
​
if __name__ == '__main__':
    for i in range(1000):
        t=Thread(target=client)
        t.start()