技术分享 2025-02-11

Python多线程(3)

Python多线程实战笔记

Python 多线程 消息总线

python多线程实战——消息总线

在项目开发过程中，总结出关于多线程的知识与经验，供开发者参考

消息总线的功能

消息总线是一个非常天才的设计，他的初衷是实现不同部件的解耦，从而实现模块化开发。

高耦合系统
以一个简单的数据流水线作为例子，假设其工作流程为：

graph TD A[数据监控模块] -->|检查数据片是否足够:不足| B[数据库读取模块] B -->|读取数据库| C[数据处理模块] C -->|处理数据为新的数据片| D[数据片更新模块]
这是一个经典的生成-消费模式，由消费者负责消耗数据片，而数据流水线作为生产者，不断监控数据片数量，提供新的数据片给消费者。

如果不使用消息总线结构，则代码实现可能如下: 

class DataMonitor:
    def __init__(self):
        # 实现参考python多线程(2)中的监控循环线程
        pass 

    def start(self):
        # 实现参考python多线程(2)中的监控循环线程
        pass 

    def stop(self):
        # 实现参考python多线程(2)中的监控循环线程
        pass 

    def is_running(self):
        # 实现参考python多线程(2)中的监控循环线程
        pass 

    def _check_data_count(self)->bool:
        # 检查数据片是否足够
        pass 

    def _read_database(self)->DataFrame:
        # 读取数据库
        pass 

    def _process_data(self, data:DataFrame)->dict:
        # 处理数据为新的数据片
        pass 

    def _update_data_count(self, new_data:dict):
        # 更新数据片
        pass 

    def monitor(self):
        # 1.检查数据片是否足够
        is_enough = self._check_data_count()

        if not is_enough:
            # 2.读取数据库
            data = self._read_database()
            # 3.处理数据为新的数据片
            new_data = self._process_data(data)
            # 4.更新数据片
            self._update_data_count(new_data)
...
将监控器用于主线程
if __name__ == "__main__":
    data_monitor = DataMonitor()
    data_monitor.start()
    ... # 其他逻辑
    data_monitor.stop()

我们发现，所有的组件都在monitor中被耦合在一起，如果需要修改某个组件，则需要修改monitor，这样会导致monitor的代码变得非常复杂，难以维护。

使用消息总线结构 使用消息总线，我们不需要显式调用各个组件，只需要将组件注册到总线，由总线负责协调组件之间的通信。

每个组件需要按照约定，调用总线提供的发布和订阅接口，来实现组件之间的通信。

graph TD %% 消息总线处于中心 BUS[消息总线] %% 组件星形分布在总线周围 A[数据监控模块] B[数据库读取模块] C[数据处理模块] D[数据片更新模块] %% 消息流：数据监控 -> 消息总线 A -->|发布:数据不足消息| BUS %% 消息总线 -> 数据库读取 BUS -->|订阅:读取数据指令| B %% 数据库读取 -> 消息总线 B -->|发布:原始数据| BUS %% 消息总线 -> 数据处理 BUS -->|订阅:处理数据指令| C %% 数据处理 -> 消息总线 C -->|发布:处理后数据| BUS %% 消息总线 -> 数据片更新 BUS -->|订阅:更新数据指令| D
这样，各个组件之间就不会相互调用了，而是通过总线进行通信。

消息总线的实现

如上所述，消息总线需要提供发布和订阅接口，来实现组件之间的通信。
此外，也可以提供类似缓存的功能，方便组件之间共享数据。

消息总线类

import threading
from typing import TypedDict, Literal, Callable

class Message(TypedDict): # 消息内部类
    message_type: Literal['data_lacking', 'data_loaded', 'data_processed']
    message_data: any

class MessageBus:

    def __init__(self):
        self.cache = {} # 数据缓存
        self.cache_lock = threading.Lock() # 缓存互斥锁
        self.subscribers = {} # 订阅注册表 *关键*  
        self.subscribers_lock = threading.Lock() # 订阅注册表互斥锁
        self.instance = None # 单例实例

    def subscribe(self, message_type: Literal['data_lacking', 'data_loaded', 'data_processed'], callback: Callable):
        """
        订阅消息  

        message_type: 消息类型
        callback: 回调函数

        订阅消息后，会根据message_type，将callback注册到subscribers中  
        当publish被调用后，会根据message_type，调用subscribers中的对应方法  
        """
        with self.subscribers_lock:
            if message_type not in self.subscribers:
                self.subscribers[message_type] = []
            self.subscribers[message_type].append(callback)

    def sync_publish(self, message:Message):
        """
        同步发布消息  
        publish被调用后，会根据message_type，调用subscribers中的对应方法，并且立即执行。会阻塞主线程  
        """
        with self.subscribers_lock:
            if message.message_type not in self.subscribers:
                return
            for callback in self.subscribers[message.message_type]:
                callback(message.message_data)

    def async_publish(self, message:Message):
        """
        异步发布消息  
        publish被调用后，会根据message_type，调用subscribers中的对应方法，并且异步执行。不会阻塞主线程  
        """
        if message.message_type not in self.subscribers:
            return
        for callback in self.subscribers[message.message_type]:
            threading.Thread(target=callback, args=(message.message_data,)).start()

    def put_cache(self, key, value):
        """
        将数据放入缓存  
        """
        with self.cache_lock:
            self.cache[key] = value

    def get_cache(self, key):
        """
        从缓存中获取数据  
        """
        with self.cache_lock:
            return self.cache.get(key)

    @staticmethod
    def get_message_bus()->'MessageBus':
        """
        获取消息总线实例  
        """
        if not MessageBus.instance:
            MessageBus.instance = MessageBus()
        return MessageBus.instance

数据监控模块

# 数据监控模块
import MessageBus # 导入消息总线

class DataMonitor:
    def __init__(self):
        self.started = False # 启动标志
        self.message_bus = MessageBus.get_message_bus() # 获取消息总线实例
        self.thread = None # 监控线程

    def start(self):
        """启动监控线程"""
        if not self.started:
            self.started = True
            self.thread = threading.Thread(target=self.monitor, daemon=True)
            self.thread.start()

    def stop(self):
        """停止监控线程"""
        self.started = False
        if self.thread and self.thread.is_alive():
            self.thread.join(timeout=5)

    def is_running(self) -> bool:
        """检查监控是否正在运行"""
        return self.started and (self.thread is None or self.thread.is_alive())
    def _check_data_count(self)->bool:
        # 检查数据片是否足够
        pass 

    def _get_df_key(self)->str:
        # 获取数据片key
        pass 

    def monitor(self):
        while self.started:
            is_enough = self._check_data_count()
            if not is_enough:
                df_key = self._get_df_key()
                message = Message(message_type='data_lacking', message_data={'df_key': df_key}) # 数据不足消息
                self.message_bus.sync_publish(message) # 发布数据不足消息

数据库读取模块

# 数据库读取模块
import MessageBus # 导入消息总线

class DatabaseLoader:
    def __init__(self):
        self.message_bus = MessageBus.get_message_bus() # 获取消息总线实例

    def load_data(self, df_key: str) -> DataFrame:
        # 根据df_key加载数据
        pass

    def data_loaded_callback(self, message_data: dict):
        # 数据加载完成回调
        df_key = message_data.get('df_key')
        df = self.load_data(df_key)
        self.message_bus.put_cache(df_key, df)

    def subscribe(self):
        # 订阅数据不足消息
        self.message_bus.subscribe('data_lacking', self.data_loaded_callback) # 订阅数据不足消息

DatabaseLoader().subscribe()  # ✅ 创建实例后调用

数据处理模块 

# 数据处理模块
import MessageBus

class DataProcessor:
    def __init__(self):
        self.message_bus = MessageBus.get_message_bus()

    def process_data(self, df: DataFrame) -> dict:
        # 处理数据为新的数据片
        pass

    def data_process_callback(self, message_data: dict):
        # 数据处理回调
        df_key = message_data.get('df_key')
        df = self.message_bus.get_cache(df_key)
        if df is not None:
            processed_data = self.process_data(df)
            self.message_bus.put_cache(f"{df_key}_processed", processed_data)

            # 发布处理完成消息
            processed_message = Message(
                message_type='data_processed',
                message_data={'df_key': df_key, 'processed_data': processed_data}
            )
            self.message_bus.async_publish(processed_message)

    def subscribe(self):
        # 订阅数据加载完成消息
        self.message_bus.subscribe('data_loaded', self.data_process_callback)

DataProcessor().subscribe()

数据片更新模块 

# 数据片更新模块
import MessageBus

class DataUpdater:
    def __init__(self):
        self.message_bus = MessageBus.get_message_bus()

    def update_data_count(self, processed_data: dict):
        # 更新数据片计数
        pass

    def data_update_callback(self, message_data: dict):
        # 数据更新回调
        processed_data = message_data.get('processed_data')
        self.update_data_count(processed_data)

    def subscribe(self):
        # 订阅数据处理完成消息
        self.message_bus.subscribe('data_processed', self.data_update_callback)

DataUpdater().subscribe()

完整使用示例 

if __name__ == "__main__":
    # 初始化所有组件
    data_monitor = DataMonitor()
    db_loader = DatabaseLoader()
    data_processor = DataProcessor()
    data_updater = DataUpdater()

    # 订阅消息
    db_loader.subscribe()
    data_processor.subscribe()
    data_updater.subscribe()

    # 启动监控
    data_monitor.start()
    print("消息总线系统已启动...")

    # 运行一段时间
    import time
    time.sleep(60)

    # 停止监控
    data_monitor.stop()
    print("消息总线系统已停止")

说明
上述实现看起来比较复杂，实际上很简单，每一个组件只是按照约定，调用消息总线提供的接口，来实现组件之间的通信。
当DataMonitor发现数据不足时，会发布数据不足消息给消息总线，它会使用sync_publish方法，发布一个data_lacking消息。sync_publish执行后，会在subscribers中找到data_lacking的回调函数，并执行。

如果是sync_publish，则需要等待回调函数执行完成，才会向后执行，实现阻塞；如果是async_publish，则不会等待回调函数执行完成，会立即向后执行。

一般来说，同步发布适用于有严格拓扑排序依赖的场景，如登录 -> 获取用户信息和token -> 使用用户token访问api，这三步需要严格顺序，不能并发执行。

而异步发布适用于没有严格拓扑排序依赖的场景，如需要预加载数据，不需要等待数据加载完成，可以立即向后执行。

其他注意事项 1.除了注册外，不同的组件需要约定好message_data的格式，以便于正确解析消息。 2.回调函数的参数必须是message_data: dict类型，因为publish方法传递的是消息数据字典。 3.需要使用单例模式，否则可能注册到不同的总线实例，导致消息无法正确传递。 4.合理选择同步发布(sync_publish)和异步发布(async_publish)：同步适用于有依赖关系的操作，异步适用于并发场景。 5.注意线程安全：消息总线内部使用锁来保证线程安全，但组件自身也需要考虑多线程安全性。