带输入的角度异步管道

import asyncio
from aio_pika import connect, Message, ExchangeType

async def main():
    connection = await connect("amqp://guest:guest@localhost/")
    channel = await connection.channel()
    exchange = await channel.declare_exchange('logs', ExchangeType.DIRECT)

    queue = await channel.declare_queue('log_queue', durable=True)
    await queue.bind(exchange, 'log')

    async with queue.iterator() as queue_iter:
        async for message in queue_iter:
            async with message.process():
                print(message.body)
                # 处理消息

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

from concurrent.futures import ThreadPoolExecutor

def process_data(data):
    # 处理数据的逻辑
    pass

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(process_data, data) for data in data_stream]
    for future in futures:
        result = future.result()

import heapq

class OrderedQueue:
    def __init__(self):
        self.queue = []
        self.index = 0

    def put(self, item, priority):
        heapq.heappush(self.queue, (priority, self.index, item))
        self.index += 1

    def get(self):
        return heapq.heappop(self.queue)[-1]

ordered_queue = OrderedQueue()

# 添加数据
ordered_queue.put("data1", 1)
ordered_queue.put("data2", 2)

# 获取数据
print(ordered_queue.get())  # 输出 "data1"
print(ordered_queue.get())  # 输出 "data2"