获取异步服务器以调用另一台服务器

Posted 2023-02-24

【中文标题】获取异步服务器以调用另一台服务器

【英文标题】：Get an asyncio server to call another server

【发布时间】：2014-10-28 06:58:55

【问题描述】：

所以我修改了 asyncio 客户端和服务器的示例（在这里找到：https://docs.python.org/3/library/asyncio-protocol.html#protocol-example-tcp-echo-server-and-client），我想要的只是让 client.py 调用 serverone.py，然后调用 servertwo.py。

client.py

#!/usr/bin/env python3.4
import asyncio

class EchoClient(asyncio.Protocol):
    message = 'This is the Client'

    def connection_made(self, transport):
        transport.write(self.message.encode())

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

    def connection_lost(self, exc):
        asyncio.get_event_loop().stop()

loop = asyncio.get_event_loop()
coro = loop.create_connection(EchoClient, '127.0.0.1', 8888)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()

serverone.py

#!/usr/bin/env python3.4

import asyncio

class EchoClient(asyncio.Protocol):
    message = 'Server One sending message'

    def connection_made(self, transport):
        transport.write(self.message.encode())

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

    def connection_lost(self, exc):
        asyncio.get_event_loop().stop()

class EchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        self.transport = transport

    def data_received(self, data):
        loop = asyncio.get_event_loop()
        coro = loop.create_connection(EchoClient, '127.0.0.1', 8889)
        loop.run_until_complete(coro)
        # close the socket
        self.transport.close()
        loop.close()

loop = asyncio.get_event_loop()
coro = loop.create_server(EchoServer, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)
try:
    loop.run_forever()
except KeyboardInterrupt:
    print("exit")
finally:
    server.close()
    loop.close()

servertwo.py

#!/usr/bin/env python3.4

import asyncio

class EchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        self.transport = transport

    def data_received(self, data):
        print('data received: '.format(data.decode()))
        self.transport.write(data)
        # close the socket
        self.transport.close()

loop = asyncio.get_event_loop()
coro = loop.create_server(EchoServer, '127.0.0.1', 8889)
server = loop.run_until_complete(coro)
try:
    loop.run_forever()
except KeyboardInterrupt:
    print("exit")
finally:
    server.close()
    loop.close()

我在终端中启动servertwo.py 和serverone.py，然后调用client.py。部分工作；客户端确实调用了调用 servertwo 的 serverone，但随后 serverone 失败并出现此错误：

Exception in callback <bound method _SelectorSocketTransport._read_ready of <asyncio.selector_events._SelectorSocketTransport object at 0x7fbf4453b048>>()
handle: Handle(<bound method _SelectorSocketTransport._read_ready of <asyncio.selector_events._SelectorSocketTransport object at 0x7fbf4453b048>>, ())
Traceback (most recent call last):
  File "/usr/lib64/python3.4/asyncio/events.py", line 39, in _run
    self._callback(*self._args)
  File "/usr/lib64/python3.4/asyncio/selector_events.py", line 458, in _read_ready
    self._protocol.data_received(data)
  File "./serverone.py", line 25, in data_received
    loop.run_until_complete(coro)
  File "/usr/lib64/python3.4/asyncio/base_events.py", line 203, in run_until_complete
    self.run_forever()
  File "/usr/lib64/python3.4/asyncio/base_events.py", line 179, in run_forever
    raise RuntimeError('Event loop is running.')
RuntimeError: Event loop is running.

文档没有涵盖很多奇怪的用例，所以我有点卡住了。我应该使用asyncio.async 拨打电话吗？我是否正确地解决了这个问题？

如何修复或避免RuntimeError？

【问题讨论】：

我很难称这是一个“奇怪的”用例。编写一个需要在服务入站请求的过程中进行出站异步调用的服务器对我来说似乎很常见。来到这里寻找有关如何做到这一点的示例。

【参考方案1】：

将asyncio.async 替换为asyncio.ensure_future，它会以赞成的答案运行。

【参考方案2】：

您可以使用asyncio.async 来调度create_connection 返回的协程由事件循环运行，然后使用asyncio.Future 的add_done_callback 方法（更具体地说，一个asyncio.Task）@协程完成后987654328@返回关闭循环：

class EchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        self.transport = transport

    def data_received(self, data):
        loop = asyncio.get_event_loop()
        coro = loop.create_connection(EchoClient, '127.0.0.1', 8890)
        fut = asyncio.async(coro)
        fut.add_done_callback(self.shutdown)

    def shutdown(self, *args):
        self.transport.close()
        loop.stop()

【讨论】：

这个答案肯定给了我一个帮助，但即使没有 shutdown() 方法，一旦运行 async 调用，服务器也会停止。我尝试在data_recieved() 方法中添加run_forever()，但无济于事。在调用另一台服务器后，我不能让服务器停止。但是对于未以异常结束的调用 +1。

我应该解释一下。我问的问题是我所面临的简化问题。我真正的问题是我需要在多个守护进程（已经分叉）之间进行通信。通过添加您的解决方案，守护程序会继续运行，我可以多次调用服务器而不会导致循环死机，这样就解决了我的问题。但是，在我的示例中，serverone.py 仍然退出其循环。

@NuclearPeon 那是因为你在EchoClient.connection_lost 中停止了事件循环。不要那样做！ :)

啊哈！我还删除了shutdown() 中的loop.stop()，这一切都按预期工作！非常感谢@dano！