非同步系統中複雜事件工作流程處理

import asyncio
from typing import AsyncIterator, List, Any

async def event_source() -> AsyncIterator[Any]:
    # 模擬一個發出不同事件的事件源
    for i in range(1, 21):
        await asyncio.sleep(0.05)
        yield {"id": i, "type": "A" if i % 2 == 0 else "B", "payload": f"data_{i}"}

async def batch_events(source: AsyncIterator[Any], batch_size: int) -> AsyncIterator[List[Any]]:
    batch = []
    async for event in source:
        batch.append(event)
        if len(batch) >= batch_size:
            yield batch
            batch = []
    if batch:
        yield batch

async def process_batch(batch: List[Any]) -> None:
    # 應用先進的過濾和聚合邏輯
    type_a_events = [e for e in batch if e["type"] == "A"]
    type_b_events = [e for e in batch if e["type"] == "B"]
    # 根據特定領域的邏輯關聯事件
    if type_a_events and type_b_events:
        print("相關批次:", type_a_events, type_b_events)
    else:
        print("無關批次:", batch)

async def orchestrate_workflow():
    async for event_batch in batch_events(event_source(), batch_size=5):
        await process_batch(event_batch)

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

import asyncio
from typing import List, Dict

async def collect_events() -> List[Dict]:
    # 模擬從非同步源收集事件
    events = []
    for i in range(10):
        await asyncio.sleep(0.1)
        events.append({"id": i, "type": "TYPE1", "data": f"payload_{i}"})
    return events

async def process_batch(events: List[Dict]) -> None:
    # 處理事件批次
    for event in events:
        print(f"Processed event: {event}")

async def main():
    events = await collect_events()
    await process_batch(events)

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

import asyncio
from collections import defaultdict
from typing import Dict, Any, Callable, List

class EventRouter:
    def __init__(self):
        self.queues: Dict[str, asyncio.Queue] = defaultdict(asyncio.Queue)
        self.handlers: Dict[str, List[Callable[[Any], None]]] = defaultdict(list)

    def register_handler(self, event_type: str, handler: Callable[[Any], None]) -> None:
        self.handlers[event_type].append(handler)

    async def dispatch_event(self, event: Dict[str, Any]) -> None:
        q = self.queues[event["type"]]
        await q.put(event)

    async def start_dispatching(self) -> None:
        tasks = []
        for event_type, queue in self.queues.items():
            tasks.append(asyncio.create_task(self._process_queue(event_type, queue)))
        await asyncio.gather(*tasks)

    async def _process_queue(self, event_type: str, queue: asyncio.Queue) -> None:
        while True:
            event = await queue.get()
            for handler in self.handlers.get(event_type, []):
                try:
                    handler(event)
                except Exception as ex:
                    print(f"Error handling event {event} in {handler}: {ex}")

# 定義事件處理程式
def sample_handler(event: Dict[str, Any]) -> None:
    print(f"Handled event: {event}")

# 非同步生成事件
async def event_generator(router: EventRouter) -> None:
    for i in range(15):
        await asyncio.sleep(0.1)
        event_type = "TYPE1" if i % 2 == 0 else "TYPE2"
        await router.dispatch_event({"id": i, "type": event_type, "data": f"payload_{i}"})

async def orchestrate_routing() -> None:
    router = EventRouter()
    router.register_handler("TYPE1", sample_handler)
    router.register_handler("TYPE2", sample_handler)
    dispatcher_task = asyncio.create_task(router.start_dispatching())
    await event_generator(router)
    await asyncio.sleep(1)
    dispatcher_task.cancel()

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

import asyncio
from typing import List, Callable, Any

async def process_batch(batch: List[Any], handler: Callable[[List[Any]], None]) -> None:
    # 以可控的方式在批次上執行處理器
    handler(batch)

async def batch_worker(queue: asyncio.Queue, batch_size: int, handler: Callable[[List[Any]], None]) -> None:
    batch = []
    while True:
        try:
            # 等待專案或在需要部分批次處理時超時
            item = await asyncio.wait_for(queue.get(), timeout=0.1)
            batch.append(item)
            if len(batch) >= batch_size:
                await process_batch(batch, handler)
                batch.clear()
        except asyncio.TimeoutError:
            if batch:
                await process_batch(batch, handler)
                batch.clear()

# 處理批次任務的示例處理器
def example_handler(batch: List[Any]) -> None:
    print("處理批次：", batch)
    # 處理批次：[item1, item2, ..., itemN]

import asyncio

async def producer(queue: asyncio.Queue) -> None:
    for i in range(1000):
        await queue.put(f"task_{i}")
        print(f"已加入 task_{i}")

async def consumer(queue: asyncio.Queue) -> None:
    while True:
        task = await queue.get()
        print(f"已消費 {task}")
        # 模擬任務處理延遲
        await asyncio.sleep(0.05)
        queue.task_done()

async def main():
    # 設定最大佇列大小以控制反壓
    queue = asyncio.Queue(maxsize=50)
    prod_task = asyncio.create_task(producer(queue))
    cons_task = asyncio.create_task(consumer(queue))
    await asyncio.gather(prod_task)
    await queue.join()
    cons_task.cancel()

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