您希望能夠盡快處理進入Reader的所有數據，但也不能讓多個線程/進程嘗試并行處理這些數據；這就是以前使用執行器遇到競爭條件的原因。相反，您應該啟動一個可以處理所有數據包數據的工作進程，一次一個，使用multiprocessing.Queue將數據從父進程傳遞給工作進程。然后，當worker已經構建、驗證并準備分發一個有效的包時，它使用另一個multiprocessing.Queue將其發送回父進程中的線程，該線程可以使用線程安全的{a1}方法來調度{}運行。在

下面是一個未經測試的示例，可以讓您了解如何做到這一點：import asyncio

import struct

from concurrent.futures.ProcessPoolExecutor

import threading

def handle_result_packets():

""" A function for handling packets to be distributed.

This function runs in a worker thread in the main process.

"""

while True:

packet = result_queue.get()

loop.call_soon_threadsafe(asyncio.async, distribute(packet))

def get_packet(): # CPU intensive

""" Handles processing all incoming packet data.

This function runs in a separate process.

"""

extra = bytearray()

while True:

data = data_queue.get()

extra.extend(data)

if len(data) > HEADER_SIZE:

payload_size, = struct.unpack_from(HEADER_FORMAT, data)

if len(data) >= HEADER_SIZE + payload_size:

packet = data[:HEADER_SIZE + payload_size]

del extra[:len(packet)]

if verify_hash(packet):

result_queue.put(packet)

class Reader(asyncio.Protocol):

def __init__(self):

self.extra = bytearray()

self.t = threading.Thread(target=handle_result_packets)

self.t.start()

def data_received(self, data):

data_queue.put(data)

if __name__ == "__main__":

loop = asyncio.get_event_loop()

data_queue = multiprocessing.Queue()

result_queue = multiprocessing.Queue()

p = multiprocessing.Process(target=get_packet)

p.start()

loop.run_until_complete(loop.create_server(Reader, '0.0.0.0', 8000))

loop.run_forever()

總結

以上是生活随笔為你收集整理的python异步和进程_Python异步缓冲区和进程d的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。