1.生产者消费者-生成器版

import time# 消费者
def consumer():cnt = yieldwhile True:if cnt <= 0:# 暂停、让出CPUcnt = yield cntcnt -= 1time.sleep(1)print('consumer consum 1 cnt. cnt =', cnt)# 生产者 (调度器)
def producer(cnt):gen = consumer()# 激活生成器next(gen)gen.send(cnt)while True:cnt += 1print('producer producer 5 cnt. cnt =', cnt)# 调度消费者current = int(time.time())if current % 5 == 0:cnt = gen.send(cnt)else:time.sleep(1)if __name__ == '__main__':producer(0)

2.生产者消费者–异步版本

import asyncio
import time
from queue import Queue
from threading import Threaddef start_loop(loop):asyncio.set_event_loop(loop)loop.run_forever()async def do_sleep(x, queue):await asyncio.sleep(x)queue.put('ok')def consumer(input_queue1, out_queue1):while True:task = input_queue1.get()if not task:time.sleep(1)continueasyncio.run_coroutine_threadsafe(do_sleep(int(task), out_queue1), new_loop)if __name__ == '__main__':print(time.ctime())new_loop = asyncio.new_event_loop()loop_thread = Thread(target=start_loop, args=(new_loop,))loop_thread.daemon = Trueloop_thread.start()input_queue = Queue()input_queue.put(5)input_queue.put(3)input_queue.put(1)out_queue = Queue()consumer_thread = Thread(target=consumer, args=(input_queue, out_queue,))consumer_thread.daemon = Trueconsumer_thread.start()while True:msg = out_queue.get()print("协程运行完...")print("当前时间:", time.ctime())

3.客户端/服务端-多线程版

客户端/服务模型

客户端

# -*- encoding=utf-8 -*-
# 客户端import socketclient = socket.socket()
print('client.fileno:', client.fileno())client.connect(('127.0.0.1', 8999))while True:content = str(input('>>>'))client.send(content.encode())content = client.recv(1024)print('client recv content:', content)

服务端

import socket
import threadingdef thread_process(s):while True:content = s.recv(1024)if len(content) == 0:breaks.send(content.upper())print(str(content, encoding='utf-8'))  # 接受来自客户端的消息，并打印出来s.close()server = socket.socket()  # 1. 新建socket
server.bind(('127.0.0.1', 8999))  # 2. 绑定IP和端口（其中127.0.0.1为本机回环IP）
server.listen(5)  # 3. 监听连接while True:s, addr = server.accept()  # 4. 接受连接new_thread = threading.Thread(target=thread_process, args=(s,))print('new thread process connect addr：{}'.format(addr))new_thread.start()

注意:

• AddressFamily=AF_INET:(用于 Internet 进程间通信)

• AddressFamily=AF_UNIX(用于同一台机器进程间通信)

• 现象:报错[WinError 10038],原因分析:socket 先 close 再调 recv 就会报错,解决办法:if not tcpCliSock._closed:

4.IO多路复用TCPServer模型

4.1Select

服务端

import select
import socket
from queue import Queue, Empty
from time import sleepserver = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.setblocking(False)
server_address = ("127.0.0.1", 8999)
print('starting up on %s port %s' % server_address)
server.bind(server_address)
server.listen(5)
inputs = [server]
outputs = []
message_queues = {}while inputs:print('waiting for the next event')readable, writable, exceptional = select.select(inputs, outputs, inputs)for s in readable:if s is server:connection, client_address = s.accept()print(f"connection from {client_address}")connection.setblocking(False)inputs.append(connection)message_queues[connection] = Queue()continuedata = s.recv(1024).decode()if data == "":print(f'closing:{s.getpeername()}')if s in outputs:outputs.remove(s)inputs.remove(s)s.close()message_queues.pop(s)continueprint(f'received {data} from {s.getpeername()} ')message_queues[s].put(data)if s not in outputs:outputs.append(s)for s in writable:try:queue_item = message_queues.get(s)send_data = ''if queue_item:send_data = queue_item.get_nowait()except Empty:print(outputs.remove(s))print(f"{s.getpeername()} has closed")else:if queue_item:s.send(send_data.encode())for s in exceptional:print(f"Exception condition on {s.getpeername}")inputs.remove(s)if s in outputs:outputs.remove(s)s.close()message_queues.pop(s)sleep(1)

客户端

import socketmessages = ['This is the message ', 'It will be sent ', 'in parts ', ]
server_address = ("127.0.0.1", 8999)
socks = [socket.socket(socket.AF_INET, socket.SOCK_STREAM), socket.socket(socket.AF_INET, socket.SOCK_STREAM), ]
print('connecting to %s port %s' % server_address)
for s in socks:s.connect(server_address)for index, message in enumerate(messages):for s in socks:print('%s: sending "%s"' % (s.getsockname(), message + str(index)))s.send((message + str(index)).encode('utf-8'))for s in socks:data = s.recv(1024)print('%s: received "%s"' % (s.getsockname(), data))if data != "":print('closing socket', s.getsockname())s.close()

• 为什么要将server放入到inputs中

在select模型中，将server放入到inputs中，当执行select时就会去检查server是否可读，就说明在缓冲区里有数据，对于server来说，有连接进入。使用accept获得客户端socket文件后，首先要放入到inputs当中，等待其发送消息。

• readable

select会将所有可读的socket返回，包括server在内，假设一个客户端socket的缓冲区里有2000字节的内容，而这一次你只是读取了1024个字节，没有关系，下一次执行select模型时，由于缓冲区里还有数据，这个客户端socket还会被放入到readable列表中。因此，在读取数据时，不必再像之前那样使用一个while循环一直读取。

• writable

在每一次写操作执行后，都从socket从writable中删除，这样做的原因很简单，该写的数据已经写完了，如果不删除，下一次select操作时，又会把他放入到writable中，可是现在已经没有数据需要写了啊，这样做没有意义，只会浪费select操作的时间，因为它要遍历outputs中的每一个socket，判断他们是否可写以决定是否将其放入到writtable中

• 异常

在exceptional中，是发生错误和异常的socket，有了这个数组，就在也不用操心错误和异常了，不然程序写起来非常的复杂，有了统一的管理，发生错误后的清理工作将变得非常简单

4.2Epoll

服务端

# -*- encoding=utf-8 -*-
# IO多路复用TCPServer模型import select
import socketdef serve():server = socket.socket()server.bind(('127.0.0.1', 8999))server.listen(1)epoll = select.epoll()epoll.register(server.fileno(), select.EPOLLIN)connections = {}contents = {}while True:events = epoll.poll(10)for fileno, event in events:if fileno == server.fileno():# 当fd为当前服务器的描述符时,获取新连接s, addr = server.accept()  # 获取套接字和地址print(f"new connection from addr:{addr},fileno:{s.fileno()},socket:{s}")epoll.register(s.fileno(), select.EPOLLIN)connections[s.fileno()] = selif event == select.EPOLLIN:# 当fd不为服务器描述符为客户端描述符时,读事件就绪，有新数据可读s = connections[fileno]content = s.recv(1024)if content:# 当客户端发送数据时print(f"recv content is {content}")print(f"fileno:{fileno} event:{event}")epoll.modify(fileno, select.EPOLLOUT)contents[fileno] = contentelse:# 当客户端退出连接时print(f"recv content is null")print(f"fileno;{fileno} event:{event} ")epoll.unregister(fileno)s.close()connections.pop(fileno)elif event == select.EPOLLOUT:# 当fd不为服务器描述符为客户端描述符时,写事件就绪try:content = contents[fileno]s = connections[fileno]s.send(content)epoll.modify(s.fileno(), select.EPOLLIN)print(f"modify content is {content}")print(f"fileno;{fileno} event:{event} ")except Exception as error:epoll.unregister(fileno)s.close()connections.pop(fileno)contents.pop(fileno)print(f"modify content is failed")print(f"fileno;{fileno} event:{event} ")if __name__ == '__main__':serve()

客户端

# -*- encoding=utf-8 -*-
# 客户端import socketclient = socket.socket()
print('client.fileno:', client.fileno())client.connect(('127.0.0.1', 8999))while True:content = str(input('>>>'))client.send(content.encode())content = client.recv(1024)print('client recv content:', content.decode())

5.异步IO多路复用TCPServer模型

import socket
import select
from collections import dequeclass Future:"""可等待对象 Future"""def __init__(self, loop):self.loop = loopself.done = Falseself.co = Nonedef set_done(self):self.done = Truedef set_coroutine(self, co):self.co = codef __await__(self):if not self.done:yield selfreturnclass SocketWrapper:"""套接字协程适配器"""def __init__(self, sock: socket.socket, loop):self.loop = loopself.sock = sockself.sock.setblocking(False)self.fileno = self.sock.fileno()def create_future_for_events(self, events):future: Future = Future(loop=self.loop)def handler():future.set_done()self.loop.unregister_handler(self.fileno)if future.co:self.loop.add_coroutine(future.co)self.loop.register_handler(self.fileno, events, handler)return futureasync def accept(self):while True:try:sock, addr = self.sock.accept()return SocketWrapper(sock, self.loop), addrexcept BlockingIOError:future = self.create_future_for_events(select.EPOLLIN)await futureasync def recv(self, backlog):while True:try:return self.sock.recv(backlog)except BlockingIOError:future = self.create_future_for_events(select.EPOLLIN)await futureasync def send(self, data):while True:try:return self.sock.send(data)except BlockingIOError:future = self.create_future_for_events(select.EPOLLOUT)await futureclass EventLoop:"""调度器:epoll事件驱动"""current = Nonerunnable = deque()epoll = select.epoll()handler = {}@classmethoddef instance(cls):if not EventLoop.current:EventLoop.current = EventLoop()return EventLoop.currentdef register_handler(self, fileno, events, handler):self.handler[fileno] = handlerself.epoll.register(fileno, events)def unregister_handler(self, fileno):self.epoll.unregister(fileno)self.handler.pop(fileno)def add_coroutine(self, co):self.runnable.append(co)def run_coroutine(self, co):try:future: Future = co.send(None)future.set_coroutine(co)except Exception as e:print(e)print('coroutine {} stopped'.format(co.__name__))def run_forever(self):while True:while self.runnable:self.run_coroutine(co=self.runnable.popleft())events = self.epoll.poll(1)for fileno, event in events:handler = self.handler.get(fileno)handler()class TCPServer:def __init__(self, loop: EventLoop):self.loop = loopself.listen_sock: SocketWrapper = self.create_listen_socket()self.loop.add_coroutine(self.serve_forever())def create_listen_socket(self, ip='localhost', port=8999):sock = socket.socket()sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)sock.bind((ip, port))sock.listen()return SocketWrapper(sock, self.loop)async def handler_client(self, sock: SocketWrapper):while True:data = await sock.recv(1024)if not data:print('client disconnected')breakawait sock.send(data.upper())async def serve_forever(self):while True:sock, addr = await self.listen_sock.accept()print(f'client connect addr = {addr}')self.loop.add_coroutine(self.handler_client(sock))if __name__ == '__main__':loop = EventLoop.instance()server = TCPServer(loop)loop.run_forever()