网络编程进阶：并发编程之协程、IO模型

from greenlet import greenlet
import time

def eat(name):
    print("%s is eating 1"%name)
    time.sleep(3)
    g2.switch("neo")   # g2 再去 switch
    print("%s is eating 2"%name)
    g2.switch("alex")  # 再 switch 到 play

def play(name):
    print("%s play 1"%name)
    g1.switch("neo")  # 再 switch 到 eat
    print("%s play 2"%name)

g1 = greenlet(eat)   # greenlet() 只能传入函数名，不要传函数参数
g2 = greenlet(play)

g1.switch("neo")  # 利用 g1.switch() 启动程序  # 此时需要传入函数的参数

"""
greenlet 并不能检测到I/O阻塞
"""

from gevent import monkey;monkey.patch_all()  # 合并成一行，专门用于打标记
import gevent
import time

def eat(name):
    print("%s is eating 1"%name)
    # gevent.sleep(3)  # gevent.sleep()和 time.sleep()效果一样
    time.sleep(3)
    print("%s is eating 2"%name)

def play(name):
    print("%s play 1"%name)
    # gevent.sleep(4)
    time.sleep(4)
    print("%s play 2"%name)

start = time.time()

g1 = gevent.spawn(eat,"neo")   # 提交任务  #  spawn()第一个参数写任务名，后面直接参数就行（位置参数或关键字参数都可以）
g2 = gevent.spawn(play,"alice")  # gevent.spawn()是异步提交任务

g1.join()
g2.join()   # 保证上面提交的两个任务都执行完毕了  # 协程是单线程的，需要再线程结束前等待g1和g2，要不然g1和g2还没起来，“主线程”就结束了，此时g1和g2也就不会再执行了

# g1.join()和g2.join()可以合并成：
# gevent.joinall([g1,g2])

"""
执行过程分析：
g1先起来，执行了第一个print，然后遇到了IO阻塞（gevent.sleep(3)),然后立马就切到了 g2 提交的 play任务，
执行 play中的第一个print，接着又遇到了IO阻塞（gevent.sleep(4)），然后就又切到了 g1的eat任务，此时g1的eat还是处于阻塞状态，接着就在两个任务之间来回切
直到 g1的eat 又处于就绪状态，打印 eat的第2个print；执行完 eat之后，g2的play还处于阻塞状态，然后等其阻塞结束后执行 play的第2个print；
通过这种方式 gevent帮你监测了多个任务之间的IO阻塞（遇到IO阻塞就切走）
"""

stop = time.time()
print(stop-start)

"""
执行时间大约是阻塞最长的时间，如果这两个任务改成串行，则执行时间就是 3+4 = 7秒多
"""

"""
gevent.sleep() 不等同于 time.sleep(),因为gevent只能识别它自己模拟的阻塞（所以，如果把gevent.sleep()改成了time.sleep(),则整个程序就还是串行）
如果想实现gevent识别所有的阻塞，就需要用到gevent模块下的monkey，monkey下面有一个功能是 patch_all(),
monkey.patch_all()的作用是把下面凡是涉及到IO操作的都会帮你打一个标记，从而能被gevent识别
但凡需要gevent模块监测一个IO操作，就需要在你整个文件的开头导入 monkey，并且做一个 monkey.patch_all()的操作（一定要在整个文件的开头写，这样文件下面多有的模块也好、功能也好，都会被做上标记）
"""

from gevent import monkey,spawn;monkey.patch_all()
from socket import *

def comm(conn):
    while True:
        try:
            data = conn.recv(1024)
            if not data: break
            conn.send(data.upper())
        except ConnectionResetError:
            conn.close()
            break

def server(ip,port):
    server = socket(AF_INET,SOCK_STREAM)
    server.bind((ip,port))
    server.listen(5)

    while True:
        conn,client_addr = server.accept()
        spawn(comm,conn)
        # 这个地方就不需要再写 .join()，因为程序能走到这一步说明server这个协程肯定已经起来了，这时线程就已经进入了while True死循环，那么comm这个协程就肯定也能起来

    server.close()

if __name__ == "__main__":
    g = spawn(server,"127.0.0.1",8080)
    g.join()  # 这个地方需要写 join()，从而保证server函数能真正起来

from socket import *
from threading import Thread,current_thread


def client():
    client = socket(AF_INET,SOCK_STREAM)
    client.connect(("127.0.0.1",8080))
    while True:
        client.send(("hello %s"%current_thread().getName()).encode("utf-8"))

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

    conn.close()

if __name__ == "__main__":
    for i in range(500):  # 模拟并发的效果
        t = Thread(target=client)
        t.start()  # 500个线程能很快起来

# 服务端

# socket套接字中的 accept,recv和send是阻塞IO，非阻塞IO模式中需要对这三个地方做处理
from socket import *

server = socket(AF_INET,SOCK_STREAM)
server.bind(("127.0.0.1",8080))
server.listen(5)
server.setblocking(False)   # Set the socket to blocking (flag is true) or non-blocking (false). # 设置socket是阻塞型还是非阻塞型；默认flag是True（阻塞型

conn_list = []
del_conn = []


while True:
    try:
        # 建链接
        conn,client_addr = server.accept()
        conn_list.append(conn)
        print(conn_list)

    except BlockingIOError:  # 如果程序在accept处遇到了IO阻塞，就执行 except
        try:
            # 接收数据、并准备发送数据的任务
            send_list = []
            for conn in conn_list:
                try:
                    data = conn.recv(1024)

                    # send也是IO，也有发生阻塞的可能性（例如发送数据过大但内存已不足），所以也需要解决这个阻塞问题
                    if not data:  # linux 系统
                        del_conn.append(conn)
                        continue
                    send_list.append((conn,data.upper()))  # 把要发送的数据及其相应的conn打包成元祖的形式添加到 send_list

                except BlockingIOError:  # 如果某个conn在recv处遇到了IO阻塞，就跳过它继续执行后面的conn
                    continue

                except ConnectionResetError:  # 如果某个conn当方面断开了链接，需要把这个conn回收，并且把它从conn_list中remove掉
                    # conn.close()
                    # 迭代对象在循环的过程中不能改变迭代对象的结构，所以此时不能把conn从conn_list中直接remove
                    del_conn.append(conn) # 把这个conn添加到要删除的列表中

            # 发消息的任务
            del_sent_list = []
            for item in send_list:  # 解决send的IO阻塞
                try:
                    conn = item[0]
                    content = item[1]
                    conn.send(content)
                    del_sent_list.append(item)  # 发送成功后，需要把这个元祖（conn，content）从 send_list中删除
                except BlockingIOError:  # 如果没发送成功
                    pass  # 循环的最后一行代码，可用pass代替continue

            # 删除已发送的信息
            for item in del_sent_list:
                send_list.remove(item)

            # 删除当方面断掉的客户端连接
            for conn in del_conn:  # 此时从connlist中删除这个conn
                conn_list.remove(conn)
        except:
            pass

"""
服务端此时就一个线程，
实现了自己去监测程序中的IO，遇到IO就切到其他任务去运行，这与gevent实现的原理类似
"""

# 客户端

from socket import *

client = socket(AF_INET,SOCK_STREAM)
client.connect(("127.0.0.1",8080))

while True:
    msg = input(">>>").strip()
    if not msg:continue
    client.send(msg.encode("utf-8"))

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

import select
from socket import *

server = socket(AF_INET,SOCK_STREAM)
server.bind(("127.0.0.1",8080))
server.listen(5)
server.setblocking(False)

rlist = [server,]  # 用于存放收消息的套接字
w1ist = []        # 用于存放发消息的套接字
wdata = {}  # 用于存放将要发送的消息

while True:
    rl,wl,xl = select.select(rlist,w1ist,[],3)  # 监测select代理的套接字
    print("rl",rl)
    print("wl",wl)

    for sock in rl:
        if sock == server:
            conn,addr = sock.accept()
            rlist.append(conn)   # 把conn这个套接字添加到 rlist这个收消息的套接字列表中
        else:
            try: # 捕捉客户端单方面断开连接的异常
                data = sock.recv(1024)
                if not data:  # linux系统
                    sock.close()
                    rlist.remove(sock)
                    continue
                w1ist.append(sock)
                wdata[sock] = data.upper()
            except Exception:
                sock.close()
                rlist.remove(sock)

    for sock in w1ist:
        data = wdata[sock]
        sock.send(data)
        wdata.pop(sock)  # 发完之后要把消息从wdata中删除
        w1ist.remove(sock)

server.close()