使用Python的asyncio.create_server实例提示用户输入

Question

我正在学习Python 3的asyncio库，并遇到了一个小问题。我试图修改Python官方文档中的EchoServer示例，使其提示用户输入而不仅仅是回显客户端发送的内容。

我认为只需要添加input()调用就可以了，但是input()会阻塞，直到有用户输入，这会导致问题。

理想情况下，即使服务器没有“话要说”，我也希望能够继续从客户端接收数据。有点像一个聊天客户端，每个连接都在与服务器聊天。我希望能够在每个单独的连接之间切换并根据需要从stdin发送输入，类似于P2P聊天客户端。

考虑以下修改后的EchoServer代码：

import asyncio

class EchoServerClientProtocol(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))

        reply = input()
        print('Send: {!r}'.format(reply))
        self.transport.write(reply.encode())

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

loop = asyncio.get_event_loop()
# Each client connection will create a new protocol instance
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)

# Serve requests until CTRL+c is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

我应该如何在服务器端从stdin获取输入，并指定要发送到哪个连接，同时仍然接收来自已连接客户端的输入？

- RG5

Answer 1

你可以使用 loop.add_reader 安排一个回调函数，在 sys.stdin 上有数据可用时运行，然后使用 asyncio.Queue 将接收到的 stdin 数据传递给你的 data_received 方法：

import sys
import asyncio


def got_stdin_data(q):
    asyncio.ensure_future(q.put(sys.stdin.readline()))

class EchoServerClientProtocol(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))
       fut = asyncio.ensure_future(q.get())
       fut.add_done_callback(self.write_reply)

   def write_reply(self, fut):
       reply = fut.result()
       print('Send: {!r}'.format(reply))
       self.transport.write(reply.encode())

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

q = asyncio.Queue()
loop = asyncio.get_event_loop()
loop.add_reader(sys.stdin, got_stdin_data, q)
# Each client connection will create a new protocol instance
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)

# Serve requests until CTRL+c is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

唯一棘手的部分是我们如何调用Queue.put/Queue.get方法；它们都是协程，无法在回调或Protocol实例方法中使用yield from调用。相反，我们只需使用asyncio.ensure_future将它们安排到事件循环中，然后使用add_done_callback方法来处理从get()调用中检索到的响应。

注意：Python 3.4.4引入了asyncio.ensure_future。在此之前，该方法称为asyncio.async。此外，Python 3.7引入了asyncio.create_task，现在是首选方法。

- dano

---

我怀疑我需要使用这样的队列，但不确定如何实现。非常感谢您的回复！它运行得很好。 - RG5

Answer 2

其他答案中的代码很好。此外，Python3.7+已经发布，我认为它可能会更简单：

   class AInput():
      def __init__(self):
          self.q = asyncio.Queue()

      async def out(self, prompt='input:'):
          print(prompt)
          await self.q.put(sys.stdin.readline().strip())

      async def input(self):
          tasks = [self.out(), self.q.get()]
          res = await asyncio.gather(*tasks)
          return res[1]


  async def demo():
      ain = AInput()
      txt = await ain.input()
      print(f"got: {txt}")

  if __name__ == "__main__":
      asyncio.run(demo())

- whi

---

很遗憾，如果在同一个线程中调用 sys.stdin.readline()，你的其他协程将无法运行，因为循环会被阻塞在等待 sys.stdin 中到达的数据。 - Arglanir