iTi-Flask/docs/MQ.md

MQ

iTi-Flask 提供通用 MQ 入口。 v1 内置 Kafka backend，底层使用 confluent-kafka。

安装

Kafka 客户端是可选依赖：

uv add "iti-flask[mq-kafka]"

未启用 MQ 的项目不需要安装 Kafka 依赖。

配置

MQ 默认关闭。

class DevConfig(BaseDevConfig):
    def __init__(self) -> None:
        super().__init__()
        self.mq_enabled = True
        self.mq = {
            "backend": "kafka",
            "bootstrap_servers": "127.0.0.1:9092",
            "group_id": "my-app",
            "client_id": "my-app-dev",
            "auto_offset_reset": "earliest",
            "failure_backoff_seconds": 1.0,
        }

也可以使用环境变量：

MQ_ENABLED=true
MQ_BACKEND=kafka
KAFKA_BOOTSTRAP_SERVERS=127.0.0.1:9092
KAFKA_GROUP_ID=my-app
KAFKA_CLIENT_ID=my-app-dev
KAFKA_AUTO_OFFSET_RESET=earliest
MQ_CONSUMER_MODE=subscribe
MQ_OFFSET_STORE_PATH=runtime/mq_offsets.sqlite
MQ_FAILURE_BACKOFF_SECONDS=1
MQ_POLL_TIMEOUT_SECONDS=1

发送消息

from iti.mq import mq_client

mq_client(app).send_json("mes.work_order.events", {"id": "MO001"}, key="MO001")
mq_client(app).send("raw.topic", value=b"bytes", key=b"k")

也可以先注册 producer：

from iti.mq import mq_registry

mq_registry.register_producer(
    name="work-order-events",
    topic="mes.work_order.events",
)

mq_client(app).sender("work-order-events").send_json({"id": "MO001"})

监听消息

使用装饰器：

from iti.mq import MQMessage, mq_consumer


@mq_consumer("mes.work_order.created", group_id="hsyh-mes")
async def handle_work_order(message: MQMessage) -> None:
    payload = message.value

默认 mode="subscribe"，依赖 Kafka consumer group。

也可以显式注册：

from iti.mq import mq_registry


def handle_work_order(message):
    payload = message.value


mq_registry.register_consumer(
    name="work-order-created",
    topics=["mes.work_order.created"],
    group_id="hsyh-mes",
    handler=handle_work_order,
)

业务模块也可以在 register_mq(app) 阶段注册 consumer 和 producer。

如果 Kafka broker 的 consumer group / offset commit 异常，也可以切到 assign 模式：

from iti.mq import mq_registry


mq_registry.register_consumer(
    name="work-order-created-direct",
    topics=["mes.work_order.created"],
    mode="assign",
    partitions="all",
    offset_store={"type": "sqlite", "path": "runtime/mq_offsets.sqlite"},
    handler=handle_work_order,
)

或者全局启用：

self.mq = {
    "backend": "kafka",
    "bootstrap_servers": "127.0.0.1:9092",
    "consumer_mode": "assign",
}

消息格式

默认 value_format="json"。 框架用 UTF-8 JSON 解码 message.value，并保留 message.raw_value。

value_format="bytes" 时，message.value 是原始 bytes。

MQMessage 包含：

• app
• topic
• partition
• offset
• key
• raw_key
• value
• raw_value
• headers
• timestamp
• raw_message

消费语义

Kafka consumer 强制设置：

enable.auto.commit = False

subscribe 模式下，handler 正常结束后框架同步提交当前消息 offset。 handler 抛错或 JSON 解码失败时，框架记录日志、不提交 offset、seek 回当前消息，并按 failure_backoff_seconds 等待后重试。

assign 模式下，框架不走 Kafka consumer group，不调用 broker offset commit。 handler 成功后把下一 offset 写入本地 offset store，默认路径是 runtime/mq_offsets.sqlite。 如果 broker 返回 offset out of range，框架按 auto_offset_reset 重置到 earliest 或 latest。

每个 consumer definition 启动一个线程，单线程顺序处理消息。

assign 模式适合当前 broker group 协议不可用的场景。 它不做自动 rebalance；多实例部署时需要业务方自己分配 partition，或者接受重复消费风险。

不支持

v1 不支持：

• DLQ。
• 批量消费。
• exactly-once 事务。
• 多 backend。
• Kafka topic 自动创建。