RocketMQ 簡(jiǎn)介
RocketMQ 是一個(gè)隊(duì)列模型的消息中間件泌霍，具有高性能甲抖，高可用钥星，高實(shí)時(shí)等特性式散，它并不支持JMS（java消息服務(wù)）規(guī)范，但參考了JMS規(guī)范和kafak等的思想打颤。
Producer 暴拄、Consumer，隊(duì)列都可以分布式编饺。
Producer可以向隊(duì)列輪流發(fā)送消息乖篷，隊(duì)列的集合稱(chēng)為T(mén)opic，Consumer可以做廣播消費(fèi)透且，也可以做集群消費(fèi)撕蔼。
能夠保證嚴(yán)格的消息順序
提供消費(fèi)者水平訂閱擴(kuò)展能力
億級(jí)的消息堆積能力
較少的依賴(lài)
RocketMQ術(shù)語(yǔ)介紹
Producer 消息生產(chǎn)者，即生產(chǎn)消息的系統(tǒng)秽誊，一般指業(yè)務(wù)系統(tǒng)鲸沮。
Consumer 消息消費(fèi)者，即消費(fèi)消息的系統(tǒng)锅论，一般指后端的異步處理系統(tǒng)讼溺。
ProducerGroup 生產(chǎn)者組，指一類(lèi)發(fā)送消息的系統(tǒng)最易。
ConsumerGroup 消費(fèi)組怒坯，指消費(fèi)同一類(lèi)消息的系統(tǒng)
Broker 消息中間件，主要完成消息的存儲(chǔ)藻懒，轉(zhuǎn)發(fā)剔猿，是消息系統(tǒng)的核心組件。
Topic 消息主題嬉荆，消息隊(duì)列的集合归敬。
MessageQueue 消息隊(duì)列，RocketMQ 的所有消息都是持久化的鄙早，可以把MessageQueue 理解為一個(gè)長(zhǎng)度無(wú)限的數(shù)組汪茧，通過(guò)下標(biāo)offset 訪問(wèn)消息
順序消息
- 消息有序指的是可以按照消息的發(fā)送順序來(lái)消費(fèi)。例如：一筆訂單產(chǎn)生了 3 條消息蝶锋，分別是訂單創(chuàng)建陆爽、訂單付款、訂單完成扳缕。消費(fèi)時(shí)，要按照順序依次消費(fèi)才有意義。與此同時(shí)多筆訂單之間又是可以并行消費(fèi)的躯舔。首先來(lái)看如下示例：
- 假如生產(chǎn)者產(chǎn)生了2條消息：M1驴剔、M2，要保證這兩條消息的順序粥庄，應(yīng)該怎樣做丧失？你腦中想到的可能是這樣
image.png
- 假定M1發(fā)送到S1，M2發(fā)送到S2惜互，如果要保證M1先于M2被消費(fèi)布讹，那么需要M1到達(dá)消費(fèi)端被消費(fèi)后，通知S2训堆，然后S2再將M2發(fā)送到消費(fèi)端描验。
- 這個(gè)模型存在的問(wèn)題是，如果M1和M2分別發(fā)送到兩臺(tái)Server上坑鱼，就不能保證M1先達(dá)到MQ集群膘流，也不能保證M1被先消費(fèi)。換個(gè)角度看鲁沥，如果M2先于M1達(dá)到MQ集群呼股，甚至M2被消費(fèi)后，M1才達(dá)到消費(fèi)端画恰，這時(shí)消息也就亂序了彭谁，說(shuō)明以上模型是不能保證消息的順序的。如何才能在MQ集群保證消息的順序允扇？一種簡(jiǎn)單的方式就是將M1马靠、M2發(fā)送到同一個(gè)Server上：
image.png
- 這樣可以保證M1先于M2到達(dá)MQServer（生產(chǎn)者等待M1發(fā)送成功后再發(fā)送M2），根據(jù)先達(dá)到先被消費(fèi)的原則蔼两，M1會(huì)先于M2被消費(fèi)甩鳄，這樣就保證了消息的順序。
- 這個(gè)模型也僅僅是理論上可以保證消息的順序额划，在實(shí)際場(chǎng)景中可能會(huì)遇到下面的問(wèn)題：
image.png
- 只要將消息從一臺(tái)服務(wù)器發(fā)往另一臺(tái)服務(wù)器妙啃，就會(huì)存在網(wǎng)絡(luò)延遲問(wèn)題。如上圖所示俊戳，如果發(fā)送M1耗時(shí)大于發(fā)送M2的耗時(shí)揖赴，那么M2就仍將被先消費(fèi)，仍然不能保證消息的順序抑胎。即使M1和M2同時(shí)到達(dá)消費(fèi)端燥滑，由于不清楚消費(fèi)端1和消費(fèi)端2的負(fù)載情況，仍然有可能出現(xiàn)M2先于M1被消費(fèi)的情況阿逃。
- 那如何解決這個(gè)問(wèn)題铭拧？將M1和M2發(fā)往同一個(gè)消費(fèi)者赃蛛，且發(fā)送M1后，需要消費(fèi)端響應(yīng)成功后才能發(fā)送M2搀菩。
- 聰明的你可能已經(jīng)想到另外的問(wèn)題：如果M1被發(fā)送到消費(fèi)端后呕臂，消費(fèi)端1沒(méi)有響應(yīng)，那是繼續(xù)發(fā)送M2呢肪跋，還是重新發(fā)送M1歧蒋？一般為了保證消息一定被消費(fèi)，肯定會(huì)選擇重發(fā)M1到另外一個(gè)消費(fèi)端2州既，就如下圖所示谜洽。
image.png
- 這樣的模型就嚴(yán)格保證消息的順序，細(xì)心的你仍然會(huì)發(fā)現(xiàn)問(wèn)題吴叶，消費(fèi)端1沒(méi)有響應(yīng)Server時(shí)有兩種情況阐虚，一種是M1確實(shí)沒(méi)有到達(dá)(數(shù)據(jù)在網(wǎng)絡(luò)傳送中丟失)，另外一種消費(fèi)端已經(jīng)消費(fèi)M1且已經(jīng)發(fā)送響應(yīng)消息晤郑，只是MQ Server端沒(méi)有收到敌呈。如果是第二種情況，重發(fā)M1造寝，就會(huì)造成M1被重復(fù)消費(fèi)磕洪。
rocketMQ本身沒(méi)有解決重復(fù)消息，如果你的業(yè)務(wù)中需要禁止重復(fù)消息需要自己在業(yè)務(wù)中去重
image.png
消息消費(fèi)者
package com.eight.rocketmq;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
* Created by h_man
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
consumer.setInstanceName("consumer_huang");
consumer.setNamesrvAddr("192.168.2.2:9876");
/**
* 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開(kāi)始消費(fèi)還是隊(duì)列尾部開(kāi)始消費(fèi)<br>
* 如果非第一次啟動(dòng)诫龙，那么按照上次消費(fèi)的位置繼續(xù)消費(fèi)
*/
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.setConsumeMessageBatchMaxSize(10);
consumer.subscribe("singleTopic_3", "TagA || TagB");
//廣播模式
//consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new MessageListenerOrderly() {
AtomicLong consumeTimes = new AtomicLong(0);
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
// 設(shè)置自動(dòng)提交
context.setAutoCommit(true);
for (MessageExt msg : msgs) {
System.out.println(msg + ",內(nèi)容：" + new String(msg.getBody()));
}
try {
TimeUnit.SECONDS.sleep(5L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.println("Consumer1 Started.");
}
}
消息提供者
package com.eight.rocketmq;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* Created by h_man
*/
public class Producer {
public static void main(String[] args) {
try {
DefaultMQProducer producer = new DefaultMQProducer("Producer");
producer.setNamesrvAddr("127.0.0.1:9876");
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
producer.start();
Message msg = new Message("singleTopic_3", //topic
"TagA", //tags
"KEY__1", //keys
(format.format(new Date()) + "<`````````11111111`>").getBytes()); //body
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, 0);
System.out.println(sendResult);
msg = new Message("singleTopic_3", //topic
"TagB", //tags
"KEY__2", //keys
(format.format(new Date()) + "<`````````2222222`>").getBytes()); //body
sendResult = producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, 0);
System.out.println(sendResult);
producer.shutdown();
} catch (MQClientException e) {
e.printStackTrace();
} catch (RemotingException e) {
e.printStackTrace();
} catch (MQBrokerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
