使用 asyncio 的 Python 简单套接字客户端/服务器

Posted 2023-03-30

【中文标题】使用 asyncio 的 Python 简单套接字客户端/服务器

【英文标题】：Python simple socket client/server using asyncio

【发布时间】：2018-01-29 17:06:41

【问题描述】：

我想使用 asyncio 协程而不是多线程来重新实现我的代码。

server.py

def handle_client(client):
    request = None
    while request != 'quit':
        request = client.recv(255).decode('utf8')
        response = cmd.run(request)
        client.send(response.encode('utf8'))
    client.close()

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('localhost', 15555))
server.listen(8)

try:
    while True:
        client, _ = server.accept()
        threading.Thread(target=handle_client, args=(client,)).start()
except KeyboardInterrupt:
    server.close()

client.py

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.connect(('localhost', 15555))
request = None

try:
    while request != 'quit':
        request = input('>> ')
        if request:
            server.send(request.encode('utf8'))
            response = server.recv(255).decode('utf8')
            print(response)
except KeyboardInterrupt:
    server.close()

我知道有一些合适的异步网络库可以做到这一点。但我只是想在这个案例上只使用 asyncio 核心库，以便更好地理解它。

在处理客户端定义之前只添加 async 关键字会非常好...这里有一段代码似乎可以工作，但我仍然对实现感到困惑。

asyncio_server.py

def handle_client(client):
    request = None
    while request != 'quit':
        request = client.recv(255).decode('utf8')
        response = cmd.run(request)
        client.send(response.encode('utf8'))
    client.close()

def run_server(server):
    client, _ = server.accept()
    handle_client(client)

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('localhost', 15555))
server.listen(8)

loop = asyncio.get_event_loop()
asyncio.async(run_server(server))
try:
    loop.run_forever()
except KeyboardInterrupt:
    server.close()

如何以最佳方式适应这一点并使用 async await 关键字。

【问题讨论】：

您是否已经完成了一些完全专注于asyncio 的教程？首先这样做可能更谨慎，而不是翻译你已经在工作的东西。我会推荐 this 让你开始。

我已经更新了这个问题。由于我仍然对 asyncio 库感到困惑，我认为这个用例可能有助于更好地理解。

如果你仍然对 asyncio 感到困惑（它确实是一堆futures、tasks 和 coroutines）并且想要避免线程，你可以尝试gevent，它是一个完善的协程库，或者，imo最好的选择，依靠操作系统的select 函数。这是 Python 3 documentation 和快速教程 Wait for IO Efficiently

【参考方案1】：

线程代码的最接近的字面翻译将像以前一样创建套接字，使其非阻塞，并使用asyncio low-level socket operations 来实现服务器。这是一个示例，坚持更相关的服务器部分（客户端是单线程的，可能原样正常）：

import asyncio, socket

async def handle_client(client):
    loop = asyncio.get_event_loop()
    request = None
    while request != 'quit':
        request = (await loop.sock_recv(client, 255)).decode('utf8')
        response = str(eval(request)) + '\n'
        await loop.sock_sendall(client, response.encode('utf8'))
    client.close()

async def run_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(('localhost', 15555))
    server.listen(8)
    server.setblocking(False)

    loop = asyncio.get_event_loop()

    while True:
        client, _ = await loop.sock_accept(server)
        loop.create_task(handle_client(client))

asyncio.run(run_server())

上述方法有效，但不是使用asyncio 的预期方式。它是非常低级的，因此容易出错，需要您记住在套接字上设置适当的标志。此外，没有缓冲，所以像从客户端读取一行这样简单的事情变得令人厌烦。此 API 级别实际上仅适用于替代事件循环的实现者，它们将提供sock_recv、sock_sendall 等的他们的实现。

Asyncio 的公共 API 提供了两个用于消费的抽象层：以 Twisted 为模型的旧的 transport/protocol layer 和较新的 streams layer。在新代码中，您几乎肯定想要使用流 API，即调用 asyncio.start_server 并避免使用原始套接字。这显着减少了行数：

import asyncio, socket

async def handle_client(reader, writer):
    request = None
    while request != 'quit':
        request = (await reader.read(255)).decode('utf8')
        response = str(eval(request)) + '\n'
        writer.write(response.encode('utf8'))
        await writer.drain()
    writer.close()

async def run_server():
    server = await asyncio.start_server(handle_client, 'localhost', 15555)
    async with server:
        await server.serve_forever()

asyncio.run(run_server())

【讨论】：

它工作得很好，即使我仍然对为什么异步和套接字看起来如此复杂以包装在同一个库中感到困惑......

@srjjio 网络（套接字）编程首先构成了需要异步 IO 的一大块动机。不支持套接字的 asyncio 库没有多大用处。

我明白了，非常感谢您的帮助@user4815162342

这是否为每个客户端保持一个套接字打开，还是为每个请求/响应创建一个新的套接字？

@DevPlayer 不，run_server 是协程函数，run_until_complete 需要等待。要创建可等待对象，您必须实际调用协程函数。这类似于期望可迭代的函数 - 您不能只向它们传递生成器函数，您必须实际调用生成器才能获得可迭代。

【参考方案2】：

我已经阅读了上面的答案和 cmets，试图弄清楚如何将asynchio lib 用于套接字。 正如 Python 经常发生的那样，官方文档和示例是有用信息的最佳来源。 我从支持文章末尾提供的示例中了解了Transports and Protocols（低级 API）和Streams（高级 API）。

例如 TCP Echo 服务器：

import asyncio


class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from '.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: !r'.format(message))

        print('Send: !r'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8888)

    async with server:
        await server.serve_forever()


asyncio.run(main())

和 TCP Echo 客户端：

import asyncio


class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        transport.write(self.message.encode())
        print('Data sent: !r'.format(self.message))

    def data_received(self, data):
        print('Data received: !r'.format(data.decode()))

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = 'Hello World!'

    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(message, on_con_lost),
        '127.0.0.1', 8888)

    # Wait until the protocol signals that the connection
    # is lost and close the transport.
    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

希望对搜索asynchio的简单解释的人有所帮助。