Python实现复杂的事件驱动架构

class EventManager:
    def __init__(self):
        self.listeners = {}

    def register_listener(self, event_type, listener):
        if event_type not in self.listeners:
            self.listeners[event_type] = []
        self.listeners[event_type].append(listener)

    def unregister_listener(self, event_type, listener):
        if event_type in self.listeners:
            self.listeners[event_type].remove(listener)

    def notify(self, event_type, data):
        if event_type in self.listeners:
            for listener in self.listeners[event_type]:
                listener(data)

# 定义事件类
class Event:
    def __init__(self, name, data=None):
        self.name = name
        self.data = data

# 监听器的简单实现
def sample_listener(event_data):
    print(f"Received event data: {event_data}")

# 创建事件管理器
event_manager = EventManager()

# 注册监听器
event_manager.register_listener('SAMPLE_EVENT', sample_listener)

# 发送事件
event_manager.notify('SAMPLE_EVENT', {'key': 'value'})

import asyncio

class AsyncEventManager:
    def __init__(self):
        self.listeners = {}

    def register_listener(self, event_type, listener):
        if event_type not in self.listeners:
            self.listeners[event_type] = []
        self.listeners[event_type].append(listener)

    async def notify(self, event_type, data):
        if event_type in self.listeners:
            tasks = []
            for listener in self.listeners[event_type]:
                if asyncio.iscoroutinefunction(listener):
                    tasks.append(listener(data))
                else:
                    listener(data)
            if tasks:
                await asyncio.gather(*tasks)

# 异步监听器实现
async def async_listener(event_data):
    await asyncio.sleep(1)  # 模拟一些异步操作，例如网络请求
    print(f"Async Received event data: {event_data}")

# 使用异步事件管理器
async_event_manager = AsyncEventManager()
async_event_manager.register_listener('ASYNC_EVENT', async_listener)

async def main():
    await async_event_manager.notify('ASYNC_EVENT', {'async_key': 'async_value'})

asyncio.run(main())

class ChainEventManager(EventManager):
    def notify(self, event_type, data):
        if event_type in self.listeners:
            for listener in self.listeners[event_type]:
                result = listener(data)
                # 检查监听器返回的数据，如果包含下一个事件，继续通知
                if isinstance(result, tuple) and len(result) == 2:
                    next_event_type, next_data = result
                    self.notify(next_event_type, next_data)

# 链式事件监听器
def first_listener(event_data):
    print(f"First listener received: {event_data}")
    return 'SECOND_EVENT', {'second_key': 'second_value'}

def second_listener(event_data):
    print(f"Second listener received: {event_data}")

# 创建链式事件管理器
chain_event_manager = ChainEventManager()
chain_event_manager.register_listener('FIRST_EVENT', first_listener)
chain_event_manager.register_listener('SECOND_EVENT', second_listener)

# 触发第一个事件
chain_event_manager.notify('FIRST_EVENT', {'first_key': 'first_value'})

pip install PyDispatcher

from pydispatch import dispatcher

# 事件信号
SOME_EVENT = 'some_event'

# 监听器函数
def handle_some_event(sender, **kwargs):
    print(f"Handled event from {sender} with data {kwargs}")

# 注册监听器
dispatcher.connect(handle_some_event, signal=SOME_EVENT, sender=dispatcher.Any)

# 发送事件
dispatcher.send(signal=SOME_EVENT, sender='main_sender', data='event_data')

pip install pika

import pika

# 连接到 RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列
channel.queue_declare(queue='event_queue')

# 生产者发送消息
def send_event(event_data):
    channel.basic_publish(exchange='',
                          routing_key='event_queue',
                          body=event_data)
    print(f"Sent event: {event_data}")

# 消费者接收消息
def on_message(ch, method, properties, body):
    print(f"Received event: {body}")

# 监听队列
channel.basic_consume(queue='event_queue', on_message_callback=on_message, auto_ack=True)

# 发送测试事件
send_event('Test Event Data')

# 开始监听
print('Waiting for events...')
channel.start_consuming()