RocketMQ學習 1.基本概念 RocketMQ是阿裡巴巴團隊使用java語言開發的一款分散式消息中間件，是一款低延遲，高可用，擁有海量消息堆積能力和靈活拓展性的消息隊列。 rocketmq的官網：http://rocketmq.apache.org gitee倉庫：https://gitee. ...
RocketMQ學習
1.基本概念
RocketMQ是阿裡巴巴團隊使用java語言開發的一款分散式消息中間件，是一款低延遲，高可用，擁有海量消息堆積能力和靈活拓展性的消息隊列。
rocketmq的官網：http://rocketmq.apache.org
gitee倉庫：https://gitee.com/apache/rocketmq?_from=gitee_search
Apache RocketMQ開發者指南：https://www.itmuch.com/books/rocketmq/
RocketMQ主要由 Producer、Broker、Consumer 三部分組成。
1.1、Producer
-
消息生產者，負責生產消息，一般由業務系統負責生產消息。
-
發送消息到Broker伺服器。
-
發送方式：
- 同步發送、
- 非同步發送、
- 順序發送、
- 單向發送。
-
同步和非同步方式均需要Broker返回確認信息，單向發送不需要。
1.2、Consumer
- 消息消費者，負責消費消息，一般是後臺系統負責非同步消費。
- 從Broker伺服器拉取消息。
- 消費形式：
- 拉取式消費、
- 推動式消費。
1.3、Broker
- Broker 在實際部署過程中對應一臺伺服器，
- 一個 Broker 可以存儲多個Topic的消息，每個Topic的消息也可以分片存儲於不同的 Broker。
- Message Queue 用於存儲消息的物理地址，每個Topic中的消息地址存儲於多個 Message Queue 中。
Topic: 消息的集合，一個主題有多條消息，每條消息只能屬於某一個主題。
Broker Server：代理伺服器，負責存儲消息、轉發消息。
Name Server：名字服務，生產者或消費者通過名字服務查找主題相應的Broker IP列表。
Message：消息，生產和消費數據的最小單位，每條消息必須屬於一個主題。
Tag：標簽，用於同一主題下區分不同類型的消息，可以根據不同業務目的在同一主題下設置不同標簽。
2、架構設計
1 技術架構
RocketMQ架構上主要分為四部分:
- Producer：支持分散式集群方式部署。
- Consumer：支持分散式集群方式部署。
- NameServer：
- 是一個非常簡單的Topic路由註冊中心，支持Broker的動態註冊與發現。
- Broker管理，NameServer接受Broker集群的註冊信息並且保存下來作為路由信息的基本數據。然後提供心跳檢測機制，檢查Broker是否還存活；
- 路由信息管理，每個NameServer將保存關於Broker集群的整個路由信息和用於客戶端查詢的隊列信息。然後Producer和Conumser通過NameServer就可以知道整個Broker集群的路由信息，從而進行消息的投遞和消費。
- BrokerServer：
- 主要負責消息的存儲、投遞和查詢以及服務高可用保證。
2 工作流程
- 啟動NameServer，NameServer起來後監聽埠，等待Broker、Producer、Consumer連上來，相當於一個路由控制中心。
- Broker啟動，跟所有的NameServer保持長連接，定時發送心跳包。心跳包中包含當前Broker信息(IP+埠等)以及存儲所有Topic信息。註冊成功後，NameServer集群中就有Topic跟Broker的映射關係。
- 收發消息前，先創建Topic，創建Topic時需要指定該Topic要存儲在哪些Broker上，也可以在發送消息時自動創建Topic。
- Producer發送消息，啟動時先跟NameServer集群中的其中一臺建立長連接，並從NameServer中獲取當前發送的Topic存在哪些Broker上，輪詢從隊列列表中選擇一個隊列，然後與隊列所在的Broker建立長連接從而向Broker發消息。
- Consumer跟Producer類似，跟其中一臺NameServer建立長連接，獲取當前訂閱Topic存在哪些Broker上，然後直接跟Broker建立連接通道，開始消費消息。
3、項目使用
官方例子：https://gitee.com/apache/rocketmq/blob/develop/docs/cn/RocketMQ_Example.md
3.1、Rocketmq下載
1、下載zip 文件解壓縮到本地磁碟中。例：D:\rocketmq
2、啟動服務，進入bin，先執行，mqnamesrv.cmd，再啟動broker，通過命令行來指定埠。
mqbroker.cmd -n localhost:9876
3.2、監控平臺下載
https://github.com/apache/rocketmq-dashboard
3.3、集成springboot項目
1、添加依賴
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.4</version>
</dependency>
2、添加配置信息
rocketmq:
name-server: http://127.0.0.1:9876 #rocketmq服務地址
producer:
group: base_group_syncMsg
send-message-timeout: 5000
retry-times-when-send-failed: 2
max-message-size: 4194304
4、測試例子
生產者
@RestController
@RequestMapping("/mq")
public class ProducerController {
@Autowired
private RocketMQTemplate mqTemplate;
@RequestMapping("/send")
public String testSend(String msg) {
try {
mqTemplate.convertAndSend("TopicTest", msg);
return "success";
} catch (Exception e) {
e.printStackTrace();
return "fail";
}
}
}
消費者
@Component
@RocketMQMessageListener(topic = "TopicTest", consumerGroup = "consumerGroupTest")
@Slf4j
public class ConsumeController implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.init("------- Consumer: {}", message);
}
}
4、冪等性
RocketMQ無法避免消息重覆（Exactly-Once），所以如果業務對消費重覆非常敏感，務必要在業務層面進行去重處理。
4.1、什麼是冪等性
對一個事情進行操作，這個操作可能會執行成百上千次，但是操作結果都是相同的，這就是冪等性。
4.2、常用解決方案
業界主流的冪等性有兩種操作：
- 唯一 ID + 指紋碼 機制，利用資料庫主鍵去重
- 缺點，每次都要查詢資料庫
- 利用redis的原子性去實現
- 缺點：可能業務執行失敗，但redis標識成功。
