Python 使用asyncio tcp

╰半夏微凉° 2023-02-08 15:59 220阅读 0赞

使用 **asyncio streams** 编写 TCP 客户端和服务端的程序

> 回显: tcp客户端发送什么，服务端就会返回什么
> 
> 本程序发送一次，客户端就会自动断开。

#### `客户端` ####

import asyncio
    
    async def tcp_echo_client(message):
        reader, writer = await asyncio.open_connection(
            '127.0.0.1', 8888)
    
        print(f'Send: {message!r}')
        writer.write(message.encode())
        await writer.drain()
    
        data = await reader.read(100)
        print(f'Received: {data.decode()!r}')
    
        print('Close the connection')
        writer.close()
        await writer.wait_closed()
    
    asyncio.run(tcp_echo_client('Hello World!'))

#### `服务端` ####

import asyncio
    
    async def handle_echo(reader, writer):
        data = await reader.read(100)
        message = data.decode()
        addr = writer.get_extra_info('peername')
    
        print(f"Received {message!r} from {addr!r}")
    
        print(f"Send: {message!r}")
        writer.write(data)
        await writer.drain()
    
        print("Close the connection")
        writer.close()
    
    async def main():
        server = await asyncio.start_server(
            handle_echo, '127.0.0.1', 8888)
    
        addr = server.sockets[0].getsockname()
        print(f'Serving on {addr}')
    
        async with server:
            await server.serve_forever()
    
    asyncio.run(main())

首先 要把代码跑起来，然后我们再了解这些asyncio stream 这些api的意思

首先来看服务端用到的API

#### <1> 服务端创建并启动一个套接字 asyncio.start\_server ####

*coroutine* asyncio.start\_server( *client\_connected\_cb* , *host=None*,*port=None*, \*\*\*, *loop=None*,

*limit=None*, *family=socket.AF\_UNSPEC*, *flags=socket.AI\_PASSIVE*,

*sock=None*, *backlog=100*, *ssl=None*, *reuse\_address=None*, *reuse\_port=None*,

*ssl\_handshake\_timeout=None*, *start\_serving=True*)

当有新的客户端 连接被建立的时候，回自动回调 *client\_connected\_cb*,该函数有2个参数，

(reader, writer)，reader是类 [`StreamReader`][StreamReader] 的实例，而writer是类 [`StreamWriter`][StreamWriter] 的实例

*client\_connected\_cb* 即可以是普通的可调用对象也可以是一个 [协程函数][Link 1]; 如果它是一个协程函数，它将自动作为 [`Task`][Task] 被调度。

*limit* 确定返回的 [`StreamReader`][StreamReader] 实例使用的缓冲区大小限制。默认情况下，*limit* 设置为 64 KiB 。

*loop* 参数是可选的。当在一个协程中await该方法时，该参数始终可以自动确定。

余下的参数将会直接传递给 [`loop.create_server()`][loop.create_server].

#### <2> 建立网络连接 asyncio.open\_connection ####

*coroutine* `asyncio.open_connection`(*host=None*, *port=None*, \*\*\*, *loop=None*,

*limit=None*, *ssl=None*, *family=0*, *proto=0*, *flags=0*, *sock=None*,

*local\_addr=None*, *server\_hostname=None*, *ssl\_handshake\_timeout=None*)

建立网络连接并返回一对 `(reader, writer)` 对象。

#### <3> reader 和 writer ####

> 然后我们再来看返回的两个类的实例 如何使用
> 
> reader是类 [`StreamReader`][StreamReader] 的实例，而writer是类 [`StreamWriter`][StreamWriter] 的实例。
> 
> 我们不推荐直接实例化这两个类，通过 [`open_connection()`][open_connection] 和 [`start_server()`][start_server]

看名字就知道了，reader 是从io流中读取数据。

*  **read**( n= -1 )
    
    n ,读取n个byte并返回
    
    默认为-1,则读取至EOF,并返回所有读取到的byte
    
    如果读取到EOF，且内部缓冲区为空，则返回空的bytes对象
 *  **readline**()
    
    读取一行，`\n` 结尾为一行。
    
    如果读到EOF而没有`\n` 则返回部分读取的数据
    
    如果读取到EOF，且内部缓冲区为空，则返回空的bytes对
 *  **readexactly**(n)
    
    精确读取n个byte，不能多也不能少
    
    如果读取到EOF，但少于n,抛出异常[`IncompleteReadError`][IncompleteReadError]
    
    可以使用 [`IncompleteReadError.partial`][IncompleteReadError.partial] 属性来获取到达流结束之前读取的 bytes 字符串
 *  `readuntil`(*separator=b’\\n’*)
    
    从流中读取数据,直到遇到指定的分隔符，返回数据包括指定的分隔符
    
    如果读取的数量，超过配置的流限制，则引发异常 [`IncompleteReadError`][IncompleteReadError] ，并重置缓冲区。
    
    [`IncompleteReadError.partial`][IncompleteReadError.partial] 属性可能包含指定separator的一部分
 *  at\_eof()
    
    如果缓冲区为空并且 `feed_eof()` 被调用，则返回 `True` 。

**writer,向io流中写入数据**

*  write(data)
    
    写入数据，一般与 drain() 一起使用
 *  writelines(data)
    
    写入的data，必须是一个字节列表，一样的 一般与 await stream.drain() 一起使用
 *  close()
    
    关闭写入器，一般与 `await stream.wait_closed()`
 *  can\_write\_eof()
    
    如果支持write\_eof 返回True
 *  write\_eof()
    
    刷新缓冲的写入数据后，关闭流的写入端
 *  **transport**
    
    Return the underlying asyncio transport.
    
    返回基础的异步传输
 *  `get_extra_info`(*name*, *default=None*)
    
    访问可选的运输信息；[`BaseTransport.get_extra_info()`][BaseTransport.get_extra_info]有关详细信息，请参见 。
 *  drain()
    
    等待，直到适合继续写入流为止
 *  `is_closing`（）
    
    如果流被关闭 或者 正在关闭，返回True
 *  `wait_closed`（）
    
    `await stream.wait_closed()` 等待流关闭

#### <4>补充Unix套接字 ####

> 操作和一般的一样，同样是返回reader和writer，只不过使用的事本机unix套接字

asyncio.**open\_unix\_connection**( *path = None*，\*\*\*，*loop = None*，

*limit = None*，*ssl = None*，*sock = None*，*server\_hostname = None*，*ssl\_handshake\_timeout = None*)

建立Unix套接字连接并返回一对 。`(reader, writer)`

与[`open_connection()`][open_connection 1]Unix套接字类似，但可在Unix套接字上运行。

asyncio.**start\_unix\_server**（*client\_connected\_cb*，*path = None*，\*\*\*，*loop = None*，

*limit = None*，*sock = None*，*backlog = 100*，*ssl = None*，*ssl\_handshake\_timeout = None*，*start\_serving = True* ）

启动Unix套接字服务器。

[`start_server()`][start_server 1]与Unix套接字类似，但可以使用。

[StreamReader]: https://docs.python.org/zh-cn/3/library/asyncio-stream.html#asyncio.StreamReader
[StreamWriter]: https://docs.python.org/zh-cn/3/library/asyncio-stream.html#asyncio.StreamWriter
[Link 1]: https://docs.python.org/zh-cn/3/library/asyncio-task.html#coroutine
[Task]: https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio.Task
[loop.create_server]: https://docs.python.org/zh-cn/3/library/asyncio-eventloop.html#asyncio.loop.create_server
[open_connection]: https://docs.python.org/zh-cn/3/library/asyncio-stream.html#asyncio.open_connection
[start_server]: https://docs.python.org/zh-cn/3/library/asyncio-stream.html#asyncio.start_server
[IncompleteReadError]: https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html#asyncio.IncompleteReadError
[IncompleteReadError.partial]: https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html#asyncio.IncompleteReadError.partial
[BaseTransport.get_extra_info]: https://docs.python.org/3/library/asyncio-protocol.html#asyncio.BaseTransport.get_extra_info
[open_connection 1]: https://docs.python.org/3/library/asyncio-stream.html#asyncio.open_connection
[start_server 1]: https://docs.python.org/3/library/asyncio-stream.html#asyncio.start_server