luc*_*uca 2 python multithreading websocket python-3.x python-asyncio
我需要创建一个软件，它同时从 web 套接字和管道接收消息，并在另一个通道上发送消息（它从套接字接收，创建一个新线程并发送到管道。以同样的方式从管道接收，创建一个新线程并发送到套接字）。
我有一个多线程问题，在程序启动时我必须启动方法socket_receiver，pipe_receiver但我只能启动pipe_receiver. 我试图消除所有的代码，并只保留socket_receiver和pipe_receiver，但它只是在进入while True的pipe_receiver。
import asyncio
import sys
import json
from concurrent.futures.thread import ThreadPoolExecutor
import websockets
# make the Pool of workers
executor = ThreadPoolExecutor(max_workers=10)
# Make connection to socket and pipe
header = {"Authorization": r"Basic XXXX="}
connection = websockets.connect('wss://XXXXXXXX', extra_headers=header)
async def socket_receiver():
"""Listening from web socket"""
async with connection as web_socket:
while True:
message = await web_socket.recv()
# send the message to the pipe in a new thread
executor.submit(send_to_pipe(message))
async def pipe_receiver():
"""Listening from pipe"""
while True:
message = sys.stdin.readline()
if not message:
break
executor.submit(send_to_socket(message))
# jsonValue = json.dump(str(line), file);
sys.stdout.flush()
def send_to_pipe(message):
# Check if message is CAM or DENM
json_message = json.loads(message)
type = int(json_message["header"]["messageID"])
# 1 is DENM message, 2 is CAM message
if type == 1 or type == 2:
# send the message to the pipe
sys.stdout.print(json_message);
async def send_to_socket(message):
async with connection as web_socket:
json_message = json.dumps(message)
await web_socket.send(json_message)
asyncio.get_event_loop().run_until_complete(
asyncio.gather(socket_receiver(),pipe_receiver()))
该程序由子进程调用，父进程通过连接到 stdout 和 stdin 的管道与其通信。
更新：我收到@Martijn Pieters 代码的异常
Traceback (most recent call last):
File "X", line 121, in <module>
main()
File "X", line 119, in main
loop.run_until_complete(asyncio.gather(socket_coro, pipe_coro))
File "X\AppData\Local\Programs\Python\Python37-32\lib\asyncio\base_events.py", line 568, in run_until_complete
return future.result()
File "X", line 92, in connect_pipe
reader, writer = await stdio()
File "X", line 53, in stdio
lambda: asyncio.StreamReaderProtocol(reader), sys.stdin)
File "X/AppData\Local\Programs\Python\Python37-32\lib\asyncio\base_events.py", line 1421, in connect_read_pipe
transport = self._make_read_pipe_transport(pipe, protocol, waiter)
File "X/AppData\Local\Programs\Python\Python37-32\lib\asyncio\base_events.py", line 433, in _make_read_pipe_transport
raise NotImplementedError
NotImplementedError
您没有ThreadPoolExecutor正确使用，而且您真的不想在这里使用它。相反，您需要设置消费者和生产者来处理带有队列的套接字和管道，以便在它们之间发送消息。
对于每种连接类型，创建一个协程来创建连接，然后将该单个连接传递给该连接的使用者和生产者任务（使用asyncio.create_task()）。使用asyncio.wait()运行有两个任务return_when=asyncio.FIRST_COMPLETED，这样你就可以取消任何当两个完成“早”之一（例如出现故障）仍在运行。
使用队列将消息从一个连接的使用者传递到另一个连接的生产者。
sys.stdin并且sys.stdout正在阻塞流，不要只是读取和写入它们！请参阅https://gist.github.com/nathan-hoad/8966377以获取尝试设置非阻塞 STDIO 流的要点，以及这个要求非阻塞流功能的asyncio 问题。
不要使用全局套接字连接，当然不要使用两个单独的async with语句。您的send_to_socket()方法实际上会关闭套接字，因为async with connection as web_socket:上下文管理器在发送第一条消息时退出，然后这会导致socket_receiver假设套接字无限期保持打开状态的代码出现问题。
不要在这里使用线程！您的连接完全由 asyncio 管理，线程将主要踩在这一点上。
asyncio.Executor()实例只能与常规可调用对象一起使用，而不应与协程一起使用。Executor.submit()声明它需要一个可调用对象，传入一个带有executor.submit(send_to_pipe(message))或executor.submit(send_to_socket(message))将导致引发异常的协程，因为协程不是可调用对象。您可能没有看到异常消息，因为该异常是在另一个线程中引发的。
这就是你的socket_receiver()协程失败的原因；它肯定会启动，但尝试发送消息失败。当我针对本地模拟 websocket 服务器运行您的代码时，会打印一条警告：
RuntimeWarning: coroutine 'send_to_socket' was never awaited
executor.submit(send_to_socket(message))
当不等待协程时，该协程中的代码永远不会执行。将协程包装在一个打印出 stderr ( try: callable(), except Exception: traceback.print_exc(file=sys.stderr)))异常的协程中，您将得到：
Traceback (most recent call last):
File "soq52219672.py", line 15, in log_exception
callable()
TypeError: 'coroutine' object is not callable
Executors 应该只用于集成不能转换为使用协程的代码；执行程序管理该代码以与asyncio任务并行运行而不受干扰。如果该代码想要与asyncio任务交互、始终使用asyncio.run_coroutine_threadsafe()或asyncio.call_soon_threadsafe()跨边界调用，则应小心。请参阅并发和多线程部分。
这是我如何重写您的代码以使用消费者/生产者模式的示例，stdio()基于主题的Nathan Hoad 要点，以及对将 stdio 视为管道的支持有限的Windows 的回退：
import asyncio
import json
import os
import sys
import websockets
async def socket_consumer(socket, outgoing):
# take messages from the web socket and push them into the queue
async for message in socket:
await outgoing.put(message)
async def socket_producer(socket, incoming):
# take messages from the queue and send them to the socket
while True:
message = await incoming.get()
jsonmessage = json.dumps(message)
await socket.send(jsonmessage)
async def connect_socket(incoming, outgoing):
header = {"Authorization": r"Basic XXXX="}
uri = 'wss://XXXXXXXX'
async with websockets.connect(uri, extra_headers=header) as websocket:
# create tasks for the consumer and producer. The asyncio loop will
# manage these independently
consumer_task = asyncio.create_task(socket_consumer(websocket, outgoing))
producer_task = asyncio.create_task(socket_producer(websocket, incoming))
# start both tasks, but have the loop return to us when one of them
# has ended. We can then cancel the remainder
done, pending = await asyncio.wait(
[consumer_task, producer_task],
return_when=asyncio.FIRST_COMPLETED
)
for task in pending:
task.cancel()
# force a result check; if there was an exception it'll be re-raised
for task in done:
task.result()
# pipe support
async def stdio(loop=None):
if loop is None:
loop = asyncio.get_running_loop()
if sys.platform == 'win32':
# no support for asyncio stdio yet on Windows, see https://bugs.python.org/issue26832
# use an executor to read from stdio and write to stdout
class Win32StdinReader:
def __init__(self):
self.stdin = sys.stdin.buffer
async def readline():
# a single call to sys.stdin.readline() is thread-safe
return await loop.run_in_executor(None, self.stdin.readline)
class Win32StdoutWriter:
def __init__(self):
self.buffer = []
self.stdout = sys.stdout.buffer
def write(self, data):
self.buffer.append(data)
async def drain(self):
data, self.buffer = self.buffer, []
# a single call to sys.stdout.writelines() is thread-safe
return await loop.run_in_executor(None, sys.stdout.writelines, data)
return Win32StdinReader(), Win32StdoutWriter()
reader = asyncio.StreamReader()
await loop.connect_read_pipe(
lambda: asyncio.StreamReaderProtocol(reader),
sys.stdin
)
writer_transport, writer_protocol = await loop.connect_write_pipe(
asyncio.streams.FlowControlMixin,
os.fdopen(sys.stdout.fileno(), 'wb')
)
writer = asyncio.streams.StreamWriter(writer_transport, writer_protocol, None, loop)
return reader, writer
async def pipe_consumer(pipereader, outgoing):
# take messages from the pipe and push them into the queue
while True:
message = await pipereader.readline()
if not message:
break
await outgoing.put(message.decode('utf8'))
async def pipe_producer(pipewriter, incoming):
# take messages from the queue and send them to the pipe
while True:
jsonmessage = await incoming.get()
message = json.loads(jsonmessage)
type = int(message.get('header', {}).get('messageID', -1))
# 1 is DENM message, 2 is CAM message
if type in {1, 2}:
pipewriter.write(jsonmessage.encode('utf8') + b'\n')
await pipewriter.drain()
async def connect_pipe(incoming, outgoing):
reader, writer = await stdio()
# create tasks for the consumer and producer. The asyncio loop will
# manage these independently
consumer_task = asyncio.create_task(pipe_consumer(reader, outgoing))
producer_task = asyncio.create_task(pipe_producer(writer, incoming))
# start both tasks, but have the loop return to us when one of them
# has ended. We can then cancel the remainder
done, pending = await asyncio.wait(
[consumer_task, producer_task],
return_when=asyncio.FIRST_COMPLETED
)
for task in pending:
task.cancel()
# force a result check; if there was an exception it'll be re-raised
for task in done:
task.result()
async def main():
pipe_to_socket = asyncio.Queue()
socket_to_pipe = asyncio.Queue()
socket_coro = connect_socket(pipe_to_socket, socket_to_pipe)
pipe_coro = connect_pipe(socket_to_pipe, pipe_to_socket)
await asyncio.gather(socket_coro, pipe_coro)
if __name__ == '__main__':
asyncio.run(main())
然后从两个任务开始，一个是管理套接字，另一个是管理 STDIO 管道。两者都为他们的消费者和生产者启动了 2 个以上的任务。有两个队列将消息从一个的消费者发送到另一个的生产者。