并发编程——协程

Posted 2020-12-25 ykgo

协程

协程：是单线程下的并发，又称微线程。英文名Coroutine。
协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。

需要强调的是：

1. python的线程属于内核级别的，即由操作系统控制调度
（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其他线程运行）
2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，
以此来提升效率（！！！非io操作的切换与效率无关）

协程实现

def consumer():
    while True:
        res = yield
        print("消费了%s"%res)


def producer(c):
    for i in range(3):
        print("生产了%s"%i)
        c.send(i)


c =consumer()   # 创建消费者
c.send(None)    # 启动消费者
producer(c)     # 生产者发送数据

c.close()      # 关闭消费者

3.1 greenlet模块

from greenlet import greenlet

def eat(name):
    print("%s is eating" % name)
    g2.switch(‘yk‘)
    print("%s is eating" % name)
    g2.switch()


def play(name):
    print("%s is playing phone" % name)
    g1.switch()
    print("%s is playing phone" % name)


if __name__ == ‘__main__‘:
    g1 = greenlet(eat)
    g2 = greenlet(play)

    g1.switch(‘yk‘) # 先让g1 开启
‘‘‘
yk is eating
yk is playing phone
yk is eating
yk is playing phone
‘‘‘

greenlet只是提供了一种比generator更加便捷的切换方式，当切到一个任务执行时如果遇到io，
那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。

3.2 gevent模块

gevent 遇到 I/O会自动切换。
```
import gevent
# 遇到 I/O自动切换

def eat(name):
    print(‘%s eat 1‘ % name)
    gevent.sleep(2) # 模拟的是gevent可以识别的io阻塞
    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, ‘dx‘) # 创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数
g2 = gevent.spawn(play, name=‘yk‘)
g1.join()
g2.join()
# 或者gevent.joinall([g1,g2])
print(‘主‘)

‘‘‘
dx eat 1
yk play 1
yk play 2
dx eat 2
主
‘‘‘

打补丁

from gevent import monkey;monkey.patch_all() # 打补丁
# 要想识别time模块，得在前面导入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(‘主‘)
```

gevent 实现并发通信

------------------------服务端-------------------
from gevent import monkey

monkey.patch_all()
import gevent

import socket


def communute(conn):
    while True:
        try:
            data = conn.recv(1024)
            if not data: break
            print("客户端数据->", data)
            conn.send(data.upper())
        except ConnectionResetError:
            break 
    conn.close()


def serve():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("localhost", 8080))
    server.listen(5)

    print("等待连接")
    while True: 
        conn, client_addr = server.accept()
        print("来自%s连接成功" % client_addr[0])
        gevent.spawn(communute, conn)

    server.close()


if __name__ == ‘__main__‘:
    serve()

---------------------客户端------------------
import socket
from threading import Thread, current_thread


def client():
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    client.connect((‘127.0.0.1‘, 8080))

    while True:
        msg = ‘%s hello‘ % current_thread()
        client.send(msg.encode(‘utf-8‘))

        data = client.recv(1024)
        print(data.decode(‘utf-8‘))

    client.close()


if __name__ == ‘__main__‘:
    for i in range(500):
        t = Thread(target=client)
        t.start()