3. 异步任务基础：消息队列（RabbitMQ/Kafka）在数据库工程中的角色与基本工作流

在数据库工程中，消息队列（Message Queue, MQ）是连接同步数据库操作与异步任务编排的核心中间件。它解耦了数据生产者和消费者，允许数据库变更事件以消息的形式被可靠地传递、缓冲和分发。本章从数据库工程师的视角，剖析 RabbitMQ 和 Kafka 的核心角色与基本工作流。

3.1 消息队列在数据库工程中的核心角色

消息队列并非简单的“通信管道”，它在数据库工程中承担着四个关键角色：

3.2 基本工作流：生产者 → 队列/主题 → 消费者

无论使用 RabbitMQ 还是 Kafka，其基本工作流都遵循“生产者-通道-消费者”模型。下图展示了数据库工程中最常见的异步消息传递路径：

3.3 RabbitMQ 与 Kafka 的架构差异与选型

RabbitMQ 和 Kafka 虽然都扮演消息队列角色，但其设计哲学和适用场景有显著差异。数据库工程师需要根据数据一致性要求、吞吐量、消息模型等因素做出选择。

3.4 数据库工程中的典型工作流：CDC + Kafka

以 MySQL binlog 捕获为例，展示 Kafka 在数据库工程中的完整工作流：

1. 捕获阶段： Debezium / Canal 作为生产者，模拟 MySQL Slave 拉取 binlog，将行变更（INSERT/UPDATE/DELETE）序列化为 Avro 或 JSON 消息。
2. 发布阶段： 消息按表名或库名路由到 Kafka Topic（如 dbserver1.inventory.customers），每个 Partition 保证同一主键的变更顺序。
3. 消费阶段： 下游消费者（如缓存同步服务、搜索引擎索引器、数仓 ETL）从指定 Partition 拉取消息，根据 offset 记录消费进度。
4. 幂等处理： 消费者在写入目标数据库或缓存时，使用业务主键或版本号做幂等去重，应对 Kafka 可能出现的重复投递。

// 伪代码：Kafka 消费者处理数据库变更事件
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        String tableName = extractTableName(record.topic());
        JsonNode changeEvent = objectMapper.readTree(record.value());
        String operation = changeEvent.get("op").asText(); // "c"=create, "u"=update, "d"=delete
        String primaryKey = changeEvent.get("after").get("id").asText();

        // 幂等写入 Redis 缓存
        if ("d".equals(operation)) {
            redisClient.del("cache:" + tableName + ":" + primaryKey);
        } else {
            redisClient.set("cache:" + tableName + ":" + primaryKey,
                            changeEvent.get("after").toString());
        }
        // 提交 offset（至少一次语义）
        consumer.commitSync();
    }
}

3.5 避坑指南：数据库工程师常见的 MQ 误区

• 误区一： 认为 MQ 能保证“恰好一次”而不做幂等。实际上，网络分区或消费者宕机重启几乎必然导致重复消息，消费者必须设计幂等逻辑。
• 误区二： 在 RabbitMQ 中盲目使用“确认模式”而不处理 nack 死信。未确认的消息会堆积在队列中，最终导致内存溢出或生产者阻塞。
• 误区三： 将 Kafka 的 offset 提交与业务事务混在一起。正确的做法是先处理业务逻辑，再提交 offset；或者使用 Kafka 事务 API 实现原子写入。
• 误区四： 忽略消息体大小。数据库变更事件可能包含大字段（如 TEXT/BLOB），过大的消息会拖慢序列化、网络传输和磁盘 I/O，建议对 payload 做裁剪或引用外部存储。