1.手動(dòng)協(xié)程操作：

# pip install geventfrom greenlet import greenletdef test(): print(’He ’) gr2.switch() # 切換到test2 print(’a ’) gr2.switch()def test2(): print(’is ’) gr1.switch() print(’student.’)gr1 = greenlet(test) # 創(chuàng)建一個(gè)協(xié)程gr2 = greenlet(test2)gr1.switch() # 切換到gr1，也就是運(yùn)行它

2. 自動(dòng)協(xié)程，遇到I/O自動(dòng)切換

import geventfrom gevent import monkey; monkey.patch_all() # 導(dǎo)入monkey，給所有支持的模塊打上補(bǔ)丁，變成非阻塞模塊，也就是使其能夠進(jìn)行協(xié)程操作def test1(n): print('I’m doing my homework.') gevent.sleep(n) # 模擬I/O操作，gevent自動(dòng)切換 print(’Keep doing my homework.’)def test2(n): print('I have no time to watch TV.') gevent.sleep(n) print(’Stop thinking!’)def test3(): print('Life is hard.')# 等待 greenlets 全部結(jié)束gevent.joinall([ gevent.spawn(test1,3), # 創(chuàng)建一個(gè)新的greeenlet對(duì)象，并規(guī)劃它去執(zhí)行 test1(3) gevent.spawn(test2,1), gevent.spawn(test3),])

結(jié)果：

I’m doing my homework.I have no time to watch TV.Life is hard.Stop thinking!Keep doing my homework.

3. socket 協(xié)程實(shí)例：

服務(wù)器：

import socketimport geventfrom gevent import monkeymonkey.patch_all() # 打補(bǔ)丁，讓所有支持的模塊變成非阻塞的方法def server(ip,port): s = socket.socket() # 生成socket對(duì)象 s.bind((ip, port)) # 綁定ip和端口 s.listen() # 設(shè)置監(jiān)聽 while True: cli, addr = s.accept() # 等待連接 gevent.spawn(handle_request, cli) # 對(duì)每一個(gè)連接，使用協(xié)程的方法生成greenletdef handle_request(conn): try: while True: data = conn.recv(1024) # 接收數(shù)據(jù) print('recv:', data) new = ’Hello ’+ data.decode().title() conn.send(new.encode(’utf8’)) # 返回?cái)?shù)據(jù) if not data:conn.shutdown(socket.SHUT_WR) except Exception as e: print(e) finally: conn.close()if __name__ == ’__main__’: server(’localhost’,9999)

客戶端：

import socketclient = socket.socket()client.connect((’localhost’,9999))while True: con = input(’>>>:’).strip() if len(con) ==0: continue client.send(con.encode(’utf-8’)) data = client.recv(1024) print(data.decode())client.close()

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。