Python多线程用于同时输入和套接字连接

Question

我已经和我的一个 friend 开始深入研究网络编码。并发和并行性是其中的重要组成部分。

我们已经创建了一个服务器和客户端来连接它们，并且效果很好。现在我们要在服务器中创建一个线程，用于检查键盘输入，同时列出套接字上的连接。也许我们得到了完全错误的东西，但我们尝试使用此代码和线程池执行，但程序卡在第一个等待调用

i = await ainput.asyncInput()

我们认为等待启动后，线程等待输入，主线程继续执行，但这似乎是错误的。

这是服务器模块:

import socket
import asyncio
import asyncron_Input as ainput

def closeServer():
    exit()    

server_address = ('localhost',6969)

async def main():

   #create TCP Socket
   serverSock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

   # prints the adress and the port of the server
   print("starting up on: ")
   print(server_address[0])
   print("port: ")
   print(server_address[1])

   # binds the socket to the given adress
   serverSock.bind(server_address)

   #listen for connection

   serverSock.listen(1)

   print("End server with 1")

   while True:
       #close server with asynco inputs
       i = await ainput.asyncInput()
       if i == "1":
           closeServer()

       #wait for connection
       print("waiting for conncetion")

       conn,client_address = serverSock.accept()
       try:
           print("connected to",client_address)
        
           while True:
               data = conn.recv(16)
               if data:
                   print("received:",data)
                   data = "succsessful"
               else:
                   print("no data")
                   break
       finally:
           #close connection
           conn.close


asyncio.run(main())

这是异步输入:

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def asyncInput():
    with ThreadPoolExecutor(1,'Async Input') as executor:
        return await asyncio.get_event_loop().run_in_executor(executor,input)

感谢您提前提供的帮助

Answer 1

您的代码有两个问题:

1. 在接受任何套接字连接之前等待输入，如果您希望代码同时发生，则不能使用 await，您需要使用 Task。
2. 您正在使用阻塞套接字。 sock.accept 和 sock.recv 默认情况下是阻塞的。它们将停止执行您的事件循环，您需要在 await 表达式中使用它们，这意味着使您的服务器套接字非阻塞，然后将它们与特殊的 asyncio 特定套接字方法一起使用。

要解决此问题，您需要在任务中包装对输入的监听，使服务器套接字非阻塞，获取正在运行的事件循环，然后使用 sock_accept 和 sock_recv 事件循环的方法。将所有这些放在一起，您的代码将如下所示:

import asyncio
import socket
from concurrent.futures import ThreadPoolExecutor

async def asyncInput():
    with ThreadPoolExecutor(1,'Async Input') as executor:
        return await asyncio.get_event_loop().run_in_executor(executor,input)

def closeServer():
    exit()

server_address = ('localhost',8000)

async def loop_for_input():
    while True:
        #close server with asynco inputs
        i = await asyncInput()
        if i == "1":
            closeServer()

async def main():

    #create TCP Socket
    serverSock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, )
    serverSock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    # prints the adress and the port of the server
    print("starting up on: ")
    print(server_address[0])
    print("port: ")
    print(server_address[1])

    # binds the socket to the given adress
    serverSock.bind(server_address)
    serverSock.setblocking(False) #make your socket non-blocking

    #listen for connection

    serverSock.listen(1)

    print("End server with 1")
    loop = asyncio.get_running_loop() # get the running event loop
    input_task = asyncio.create_task(loop_for_input()) # create an task to run your input loop
    while True:
        #wait for connection
        print("waiting for conncetion")

        conn,client_address = await loop.sock_accept(serverSock) # use the sock_accept coroutine to asynchronously listen for connections
        try:
            print("connected to",client_address)

            while True: # you may also want to create a task for this loop.
                data = await loop.sock_recv(conn, 16) # use the sock_recv coroutine to asynchronously listen for data
                if data:
                    print("received:",data)
                    data = "succsessful"
                else:
                    print("no data")
                    break
        finally:
        #close connection
            conn.close()


asyncio.run(main())

可能存在第三个问题，因为您的代码一次只能处理一个客户端，因为您进入第一个连接的无限循环。这意味着任何其他连接的客户端都将被阻止。如果您想解决这个问题，只要客户端连接，就创建一个新的 Task 来监听来自客户端的数据，类似于上面的代码对 asyncInput() 所做的操作>