开发 Funboost Mixin 扩展

概述

Mixin 可以向任意 funboost broker 添加横切关注点（监控、链路追踪、限流等），而无需修改 broker 本身。利用 Python 的 MRO 拦截方法。

核心原则： Mixin 重写特定方法，调用 super() 继续链路，配置存在 user_options 中。

适用场景

• 为任务执行添加监控/指标
• 实现熔断器、限流器
• 添加 OpenTelemetry / Prometheus 埋点
• 自定义任务执行前后的处理逻辑
• 通过 MRO 组合多种行为

已有 Mixin 参考

Mixin	用途	位置
CircuitBreakerConsumerMixin	高错误率时停止消费	funboost/contrib/override_publisher_consumer_cls/
MicroBatchConsumerMixin	微批消费（攒多条再处理）	funboost/contrib/override_publisher_consumer_cls/
PrometheusConsumerMixin	导出 Prometheus 指标	funboost/contrib/override_publisher_consumer_cls/
PrometheusPublisherMixin	发布端 Prometheus 指标	funboost/contrib/override_publisher_consumer_cls/
AutoOtelConsumerMixin	OpenTelemetry 链路追踪	funboost/contrib/override_publisher_consumer_cls/
AutoOtelPublisherMixin	发布端 OpenTelemetry 追踪	funboost/contrib/override_publisher_consumer_cls/
PeriodicQuotaConsumerMixin	时间窗口配额限制	funboost/contrib/override_publisher_consumer_cls/
AlertNotifierConsumerMixin	异常告警通知	funboost/contrib/override_publisher_consumer_cls/

Mixin 模板

from funboost.consumers.base_consumer import AbstractConsumer

class MyConsumerMixin(AbstractConsumer):
    """
    继承 AbstractConsumer 是可选的（仅为 IDE 自动补全）。
    不继承也能正常工作——运行时 mixin 通过动态多重继承合并 MRO。
    """

    def custom_init(self):
        super().custom_init()
        # 从 user_options 读取配置
        opts = self.consumer_params.user_options.get("my_mixin_options", {})
        self._threshold = opts.get("threshold", 10)
        self._counter = 0

    def _submit_task(self, kw):
        """任务提交到线程池前的前置检查/限流"""
        if self._counter > self._threshold:
            print(f"[WARN] 已提交 {self._counter} 次，超过阈值 {self._threshold}，可进行限流")
        super()._submit_task(kw)

    def _both_sync_and_aio_frame_custom_record_process_info_func(
        self, current_function_result_status, kw
    ):
        """
        每个任务执行后调用（同步和异步都会触发）。
        注意：此钩子内禁止有 IO 阻塞操作（如 HTTP 请求、数据库写入），
        否则会拖慢消费速度。如需 IO 操作，请用 _frame_custom_record_process_info_func
        （仅同步触发）或异步钩子配合 simple_run_in_executor。
        """
        super()._both_sync_and_aio_frame_custom_record_process_info_func(
            current_function_result_status, kw
        )
        if current_function_result_status.success:
            self._counter = 0  # 成功时重置计数
        else:
            self._counter += 1

使用方式

from funboost import boost, BoosterParams, BrokerEnum

@boost(BoosterParams(
    queue_name="monitored_task",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    consumer_override_cls=MyConsumerMixin,
    user_options={
        "my_mixin_options": {
            "threshold": 100,
        }
    },
))
def my_task(x):
    return x * 2

关键重写点

前置钩子

方法	调用时机	用途
_submit_task(kw)	任务进入线程池前	限流、熔断、配额检查
_before_start_consuming_message_hook()	消费者启动时（一次性）	初始化连接、注册指标

后置钩子

方法	调用时机	用途
_both_sync_and_aio_frame_custom_record_process_info_func	每个任务执行后（仅 CPU 操作）	计数器、状态更新
_frame_custom_record_process_info_func	每个任务执行后（同步线程中，允许 IO）	写数据库、HTTP 调用
_aio_frame_custom_record_process_info_func	每个任务执行后（异步协程中，允许 IO）	异步写数据库

三种后置钩子的选择原则：

• _both_sync_and_aio_frame_custom_record_process_info_func(self, current_function_result_status, kw) 禁止 IO 阻塞。在任务框架的核心路径上调用（同步/异步统一入口），适合纯内存操作（计数器递增、状态标记）。

• _frame_custom_record_process_info_func(self, current_function_result_status, kw) 允许同步 IO。在同步消费模式（threading/gevent/eventlet）下的并发池工作线程中调用（与任务执行同线程），适合写数据库、发 HTTP 请求等阻塞操作。不会阻塞拉取消息的调度线程，但长时间 IO 会占用 worker 降低有效并发。

• _aio_frame_custom_record_process_info_func(self, current_function_result_status, kw) 允许异步 IO。在异步消费模式（concurrent_mode=ASYNC）下作为协程调用，适合 await 异步数据库写入。

class MonitorMixin(AbstractConsumer):
    def custom_init(self):
        super().custom_init()
        self._total = 0
        self._failures = 0

    def _both_sync_and_aio_frame_custom_record_process_info_func(
        self, current_function_result_status, kw
    ):
        """纯内存操作 — 禁止 IO"""
        super()._both_sync_and_aio_frame_custom_record_process_info_func(
            current_function_result_status, kw
        )
        self._total += 1
        if not current_function_result_status.success:
            self._failures += 1

    def _frame_custom_record_process_info_func(
        self, current_function_result_status, kw
    ):
        """同步 IO 允许 — 如写数据库、发告警"""
        super()._frame_custom_record_process_info_func(
            current_function_result_status, kw
        )
        if not current_function_result_status.success:
            import requests
            requests.post("http://alert.example.com/notify", json={
                "queue": self.queue_name,
                "error": str(current_function_result_status.exception),
                "total": self._total,
            })

    async def _aio_frame_custom_record_process_info_func(
        self, current_function_result_status, kw
    ):
        """异步 IO 允许 — 异步消费模式下使用"""
        await super()._aio_frame_custom_record_process_info_func(
            current_function_result_status, kw
        )
        if not current_function_result_status.success:
            import aiohttp
            async with aiohttp.ClientSession() as session:
                await session.post("http://alert.example.com/notify", json={
                    "queue": self.queue_name,
                    "error": str(current_function_result_status.exception),
                })

执行包裹

方法	调用时机	用途
_run(kw)	同步任务执行	链路追踪 Span、计时
_async_run(kw)	异步任务执行	异步链路追踪

配置规范

将 mixin 专属配置放在 user_options 的命名空间 key 下：

user_options={
    "circuit_breaker_options": {
        "failure_threshold": 5,
        "recovery_timeout": 60,
    },
    "my_custom_mixin_options": {
        "my_key": "my_value",
    },
}

这样防止多个 mixin 的配置键冲突。

组合多个 Mixin

class CombinedMixin(CircuitBreakerConsumerMixin, PrometheusConsumerMixin):
    """MRO 确保两个 mixin 的方法都能执行"""
    pass

@boost(BoosterParams(
    queue_name="combined_task",
    consumer_override_cls=CombinedMixin,
    user_options={
        "circuit_breaker_options": {"failure_threshold": 5},
    },
))
def my_task(x): ...

两个 mixin 都能正常工作，因为它们都调用了 super() — MRO 正确串联。

异步兼容

如果后置钩子涉及 IO 操作：

from funboost.concurrent_pool.async_helper import simple_run_in_executor

class MyAsyncAwareMixin(AbstractConsumer):

    def _frame_custom_record_process_info_func(self, status, kw):
        """同步版本（含 IO）"""
        super()._frame_custom_record_process_info_func(status, kw)
        self._save_to_db(status)

    async def _aio_frame_custom_record_process_info_func(self, status, kw):
        """异步版本 — 通过 executor 复用同步逻辑"""
        await super()._aio_frame_custom_record_process_info_func(status, kw)
        await simple_run_in_executor(
            self._frame_custom_record_process_info_func, status, kw
        )

常见错误

错误	修正
重写时忘记调 super()	除抽象方法外始终调用 super()
user_options 键名扁平化	应放在 mixin 专属的嵌套 key 下
只重写 _run 不重写 _async_run	必须同时处理同步和异步路径
到处写 try/except 防御	让异常正常抛出——funboost 处理重试
不继承导致无 IDE 补全	继承 AbstractConsumer 获取自动补全（运行时可选）
多 mixin 组合时 MRO 冲突	确保所有 mixin 中一致调用 super()

Publisher Mixin 示例

from funboost.publishers.base_publisher import AbstractPublisher

class MyPublisherMixin(AbstractPublisher):

    def _publish_impl(self, msg: str):
        """拦截发布，添加日志/指标"""
        self._log_publish(msg)
        super()._publish_impl(msg)

    def _after_publish(self, publish_msg_context):
        super()._after_publish(publish_msg_context)
        self._record_publish_metric()

参考代码位置

• 已有 mixin：funboost/contrib/override_publisher_consumer_cls/
• 基础 consumer 类：funboost/consumers/base_consumer.py
• 基础 publisher 类：funboost/publishers/base_publisher.py
• 教程：funboost_all_docs_and_codes.md 4.21b 章节

相关 Skill

• developing-funboost-broker — 新增消息中间件
• funboost-observability — 监控、链路追踪与告警