python中使用asyncio的客户端访问服务器实例

注：1.本文示例代码均为《Python3标准库》与《python3.9.6帮助文档》中转载、改编
2.本文代码均在python3.9.6下测试，且使用了3.7以上版本新增的函数

这篇文章通过一组示例代码深入探究，详细的解释了asyncio中使用Protocol类让客户端与服务器通信的操作方法。（正文就是代码部分，内容写在注释里面）
这个例子没有太多体现协程的并发特性，主要是为了展现Protocol与Transport类的用法

基础知识：
服务器主体就是用户继承的asyncio.Protocol类，该类主要有4个函数可供重写：
1.connection_made(transport)：连接上时调用
2.data_received(data)：收到数据时调用
3.eof_received()：收到EOF（对方调用write_eof()）时调用
4.connection_lost(error)：连接断开时调用（发生异常或者连接结束）
都是回调函数
调用顺序：1 -> 0~n次2 -> 0或1次3 -> 4

服务器代码完整版：

import asyncio
import logging
import sys

SERVER_ADDRESS = ('localhost', 10000)  # 自定义的主机、端口

logging.basicConfig(level=logging.DEBUG,
                    format='%(name)s: %(message)s',
                    stream=sys.stderr)  # 配置log用于显示消息（更正式，比print更好，可以收到来自asyncio模块的消息）
log = logging.getLogger('main')


class EchoServer(asyncio.Protocol):
    """一直监听指定的端口，并在连接后把对方发来的东西再发回去"""
    # 甚至可以不用定义__init__()
    def connection_made(self, transport: asyncio.BaseTransport) -> None:
        """新客户连接时调用该方法，需要自己定义建立一个新的连接所需的资源"""
        self.transport = transport  # python给这个连接包装的一个接口（详细信息见注释1）
        self.address = transport.get_extra_info('peername')  # 获取信息，即(hostaddr, port)二元元组（该函数用法见注释2）
        self.log = logging.getLogger('EchoServer_{}_{}'.format(*self.address))
        self.log.debug('connection accepted')

    def data_received(self, data: bytes) -> None:
        """从客户端向服务器发送数据，服务器收到后调用，用于处理收到的数据（主要过程都在这）"""
        self.log.debug(f'received {data!r}')
        self.transport.write(data)
        self.log.debug(f'sent {data!r}')

    def eof_received(self) -> None:
        """当对方调用write_eof()时调用，似乎没啥用（删了也行）。。。应该是应付一些必须要传EOF的对象的"""
        self.log.debug('received EOF')
        if self.transport.can_write_eof():  # 检测对方是否接受write_eof()
            self.transport.write_eof()
            # 回送一个write_eof(), 给对面传递关闭信号
            # 这里不需要关闭

    def connection_lost(self, error) -> None:
        """连接发生错误或断开连接时调用，有错就传到error参数，如果只是断开连接error参数为None"""
        if error:  # 若要制造错误，可以让client中途强行关闭（还有什么方法吗）
            self.log.debug(f'ERROR: {error}')
        else:
            self.log.debug('closing')
        super().connection_lost(error)  # 可以让父类处理一些基本的问题


async def main():
    # 获取事件循环的引用，因为我们后面将要用到底层的API
    event_loop = asyncio.get_running_loop()
    # 若在event_loop 还没有跑起来时就要获得它，需要用get_event_loop()

    # 这个函数把你的服务器放进事件循环让它监听指定的端口，并返回一个Server对象（Server对象属性见注释3）
    server = await event_loop.create_server(EchoServer, *SERVER_ADDRESS,)

    log.debug(f'starting up on {SERVER_ADDRESS[0]} port {SERVER_ADDRESS[1]}')

    # async with 语句保证其中的代码块结束后server不再接受任何连接申请
    # 此处去掉似乎没事？
    async with server:
        await server.serve_forever()


asyncio.run(main())

（注释感兴趣的同学可以看看）

注释1：Transport常用属性（都是字面意思）
close()
get_extra_info(name, default=None)
is_reading()-> Bool
pause_reading()
resume_reading()
can_write_eof()
write_eof()
write(data)
writelines(list_of_data)

注释2：get_extra_info(name:str)详解
Transport.get_extra_info(name:str)可查询信息类别:
套接字:
'peername': 套接字链接时的远端地址，socket.socket.getpeername() 方法的结果 (出错时为 None ) 即(hostaddr, port)二元元组
'socket': socket.socket 实例
'sockname': 套接字本地地址， socket.socket.getsockname() 方法的结果
SSL套接字:
'compression': 用字符串指定压缩算法，或者链接没有压缩时为 None ；ssl.SSLSocket.compression() 的结果。
'cipher': 一个三值的元组，包含使用密码的名称，定义使用的SSL协议的版本，使用的加密位数。 ssl.SSLSocket.cipher() 的结果。
'peercert': 远端凭证； ssl.SSLSocket.getpeercert() 结果。
'sslcontext': ssl.SSLContext 实例
'ssl_object': ssl.SSLObject 或 ssl.SSLSocket 实例
管道：
'pipe': 管道对象
子进程：
'subprocess': subprocess.Popen 实例

注释3：Server常用属性
async with 句式
coroutine start_serving()（前面加coroutine 代表这是个协程）
coroutine serve_forever()
is_serving()
coroutine wait_closed() 等待关闭
sockets 返回当前所有连接到的socket的列表的副本

客户端代码完整版：

import asyncio
import logging
import sys


# python3的普通字符串要传输的话需要先编码，这里使用bytes
# 分三段，但拼起来是一整句话，空格也衔接上了
MESSAGES = [b'This is the message, ',
            b'it will be sent ',
            b'in parts.', ]

# 要访问的地址
SERVER_ADDRESS = ('localhost', 10000)

# 相同的配置
logging.basicConfig(level=logging.DEBUG,
                    format='%(name)s: %(message)s',
                    stream=sys.stderr)


class EchoClient(asyncio.Protocol):
    def __init__(self,future,messages):
        super().__init__()
        self.messages = messages  # 要传的信息
        self.f = future  # 指示状态用

    def connection_made(self, transport) -> None:
        """asyncio.create_connection()会调用这个方法以开始连接。同样要在这里完成客户端主要的事情"""
        self.transport = transport
        self.address = transport.get_extra_info('peername')
        self.log = logging.getLogger('EchoClient_{}_{}'.format(*self.address))
        self.log.debug('connecting to {} port {}'.format(*self.address))

        # 看上去是分步发送消息，实际上底层网络代码可能会把多个消息结合到一起一次传输
        for msg in self.messages:
            transport.write(msg)
            self.log.debug(f'sending {msg!r}')
        # transport.writelines(self.messages)  # 如果不需分步显示，则可以这样改进

        # 作为结束,不然停不下来
        if transport.can_write_eof():
            transport.write_eof()

    def data_received(self, data) -> None:
        self.log.debug(f'received {data!r}')

    def eof_received(self):
        """把这个函数删掉以后还是会正常结束（帮助文档里连这个函数都没写）"""
        self.log.debug('received EOF')
        self.transport.close()

    def connection_lost(self, exc) -> None:
        self.log.debug('server closed connection')
        self.transport.close()
        super().connection_lost(exc)  # 同样将exc交给父类处理
        self.f.set_result(True)  # 最后设置future为完成


async def main():
    log = logging.getLogger('main')
    event_loop = asyncio.get_running_loop()
    future = event_loop.create_future()  # 完成状态的指示器
    log.debug('waiting for client to complete')
    # 启动客户端，lambda传参
    await event_loop.create_connection(
        lambda: EchoClient(future, MESSAGES), *SERVER_ADDRESS)
    # 等待future被设为完成
    await future


asyncio.run(main())

总结：
通过protocol类通信的客户端与服务器区别主要在于启动方式不同，其实继承出来的Protocol 类都遵循相同的规则，只要掌握四个函数的继承与分工，就能掌握Protocol类。
启动后的client/server都不会占用main函数，可以同时做其他事情，但也因此需要一个future来指示任务的完成。

本人写这篇文章也是出于兴趣，自己摸索着写出来的，参考了很多资料，做了很多测试。虽然还有很多不足之处，但是基本的知识应该讲清楚了，不熟悉的话建议多看几遍，毕竟主体是代码。我自身也在写文章的过程中收获了很多知识，认识到了什么叫教学相长。
2021.8.4