From 132d2a5ab5adf5877522e03a10dc4ff43c65ca0c Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sun, 24 May 2026 21:02:18 +0800 Subject: [PATCH] vault backup: 2026-05-24 21:02:18 --- hhs/MQ/02-消息模型/3-MQ-消息模型.md | 89 ++++++++++++++++++++++++++++- 1 file changed, 86 insertions(+), 3 deletions(-) diff --git a/hhs/MQ/02-消息模型/3-MQ-消息模型.md b/hhs/MQ/02-消息模型/3-MQ-消息模型.md index a07bc06..a8eda10 100644 --- a/hhs/MQ/02-消息模型/3-MQ-消息模型.md +++ b/hhs/MQ/02-消息模型/3-MQ-消息模型.md @@ -1,6 +1,9 @@ --- tags: - MQ + - 消息模型 + - ConsumerGroup + - 发布订阅 create time: 2026-05-24 19:52 --- @@ -135,15 +138,90 @@ RabbitMQ 使用 Exchange(交换器)实现灵活的消息路由: - Exchange 根据 Binding Rule(绑定规则)将消息路由到一个或多个 Queue - Broker 主动将消息推送给订阅了 Queue 的 Consumer -Exchange 的四种类型:Direct(精确匹配)、Topic(通配符匹配)、Fanout(广播)、Headers(头部匹配)。 +```mermaid +graph LR + P["Producer"] -->|"routing_key = order.created"| EX["Topic Exchange"] + EX -->|"order.created"| QA["Queue: order-service"] + EX -->|"order.created"| QB["Queue: audit-log"] + EX -->|"order.*"| QC["Queue: analytics"] + QA --> CA["Order Consumer"] + QB --> CB["Audit Consumer"] + QC --> CC["Analytics Consumer"] + + style EX fill:#4A90D9,color:#fff +``` + +一张消息进入 Exchange 后,可以被路由到多个 Queue——这意味着 RabbitMQ 天然支持发布订阅,不需要额外的 Consumer Group 概念。 + +Exchange 的四种类型:**Direct**(精确匹配 Routing Key)、**Topic**(通配符匹配,`*` 匹配一个词,`#` 匹配多个词)、**Fanout**(忽略 Routing Key,广播到所有绑定队列)、**Headers**(按消息头部属性匹配)。实际开发中 Topic 和 Direct 最常用。 + +```go +// RabbitMQ: 通过 Exchange 实现发布订阅 +// 生产者: 发送到 Exchange,而非直接发送到 Queue +ch.ExchangeDeclare("order_events", "topic", true, false, false, false, nil) +ch.Publish("order_events", "order.created", false, false, amqp.Publishing{ + Body: orderJSON, +}) + +// 消费者 A: 精确订阅 "order.created" +ch.QueueBind("order-queue", "order.created", "order_events", false, false, nil) +msgs, _ := ch.Consume("order-queue", "", false, false, false, false, nil) +for msg := range msgs { + processOrder(msg.Body) + msg.Ack(false) // 手动确认 +} + +// 消费者 B: 通配符订阅所有 order 事件 +ch.QueueBind("analytics-queue", "order.*", "order_events", false, false, nil) +``` + +> [!question] +> RabbitMQ 没有 Consumer Group 概念,它怎么实现"同一条消息在组内只消费一次"的语义? #### RocketMQ:ConsumerGroup + Pull RocketMQ 的模型介于 Kafka 和 RabbitMQ 之间: - 采用 ConsumerGroup 概念,与 Kafka 类似 - 默认使用长轮询 Pull 模式(Push 的底层实现其实是 Pull 封装) +- 支持集群消费(Clustering)和广播消费(Broadcasting)两种模式 - 支持顺序消费和延迟消息等高级特性 +```go +// RocketMQ 消费者: 集群消费模式 +// 同一 ConsumerGroup 内的实例分摊消息 +c, _ := rocketmq.NewPushConsumer( + consumer.WithGroupName("order-service-group"), + consumer.WithNameServer([]string{"127.0.0.1:9876"}), +) +// 订阅 Topic, * 表示不过滤 Tag +c.Subscribe("order_created", consumer.MessageSelector{}, + func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) { + for _, msg := range msgs { + processOrder(msg.Body) + } + return consumer.ConsumeSuccess, nil + }, +) +c.Start() +``` + +> [!question] +> RocketMQ 同时支持集群消费和广播消费——什么时候该用广播模式?想想"本地缓存刷新"这个场景。 + +#### 三大 MQ 消息模型速查 + +| 维度 | Kafka | RabbitMQ | RocketMQ | +|------|-------|----------|----------| +| 基本模型 | 发布订阅(Consumer Group) | 发布订阅(Exchange + Queue) | 发布订阅(Consumer Group) | +| 消费模式 | Pull(长轮询) | Push(Broker 推送) | Pull(长轮询,Push 是封装) | +| 并行单元 | Partition | Queue | MessageQueue | +| 分组机制 | Consumer Group | 无原生分组(用多个 Queue 模拟) | ConsumerGroup | +| 路由能力 | 基于 Partition Key | 极强(Exchange 四种类型) | 基于 Tag / SQL92 过滤 | +| 消息回溯 | 支持(Offset 重置) | 不支持(消费即删除) | 支持(按时间戳回溯) | + +> [!tip] +> 选型时别只看吞吐量——**路由灵活性**、**消息回溯**、**消费模式**这些维度往往更影响日常开发体验。详见 [[27-MQ-选型对比]]。 + ### Go 代码示例:消费者组消费逻辑 下面是一个简化的消费者组实现,演示了 Partition 分配和消费的核心逻辑: @@ -206,5 +284,10 @@ Kafka 选择 Pull 模式有三个核心考量: ## 关联笔记 -- [[1-MQ-基础概念]] -- [[2-MQ-适用场景与选型原则]] +- [[1-MQ-基础概念]] — MQ 核心术语与基础架构 +- [[2-MQ-适用场景与选型原则]] — 何时需要 MQ、选型决策框架 +- [[22-Kafka]] — Kafka 架构、存储与生态深度剖析 +- [[23-RabbitMQ]] — RabbitMQ 的 Exchange 类型、高级特性与集群模式 +- [[24-RocketMQ]] — RocketMQ 事务消息、延迟消息与 5.0 新特性 +- [[27-MQ-选型对比]] — 七大维度横向对比各主流 MQ +- [[12-MQ-消息确认与持久化]] — ACK 机制如何保证消息不丢