1、RocketMQ的系统架构
1、Producer (生产者)
RocketMQ中的消息生产者都是以生产组（Producer Group）的形式出现的，生产组是同一类生产者的集合。
这类Producer发送相同的Topic类型的消息
2、Consumer(消费者)
RocketMQ中的消息消费者都是以消费组（Consumer Group）的形式出现的，消费组是同一类消费者的集合。
这类Consumer消费的是同一个Topic类型的消息，消费组使得在消息消费方面，
实现**负载均衡**(将一个Topic中的不同的Queue平均分配给同一个Consumer Group 中不同的Consumer)和
**容错**(一个Consumer挂了，该Consumer Group 中的其它Consumer可以接着消费原Consumer消费的Queue)。
3、Name Server(Broker与Topic路由的注册中心，支持Broker的动态注册与发现)
（1）Broker管理： 接受Broker集群的注册信息并且保存下来作为路由信息的基本数据：提供心跳检测机制，检查Broker是否还存活。
（2）路由信息管理： 每个NameServer中都保存着Broker集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer可以获取整个Broker集群的路由信息，从而进行消息的投递和消费
（3）路由注册：
NameServer通常也是以集群的方式部署，不过NameServer是无状态的，NameServer集群中的各个节点间是无差异的，各节点间相互不进行信息通讯，那各节点中的数据是如何进行数据同步的呢？再Broker节点启动时，轮询与每个NameServer节点建立长连接，发送注册请求，在NameServer内部维护着一个Broker列表，用来动态存储Broker的信息。（NameServer扩容是需要哦配置NameServer相关的信息不然Broker的数据同步不到新的NameServer上）Broker节点会每30秒发送一次心跳至NameServer，心跳包含BrokerId、Broker地址、Broker名称、Broker所属集群名称，NameServer收到心跳包会更新心跳时间戳记录Broker的最新存活时间。
（4）路由剔除：
当Broker宕机了，NameServer接受Broker心跳时没有接收到，会进行删除掉当前没有接收到心跳的Broker
（5）路由发现：
1、Push模型: 当Name Server的Topic路由数据发送变化时会主动推送消息给客户端。2、Pull模型：客户端会每隔30秒向NameServer拉取一次路由数据。3、LongPolling模型：长轮询模型，客户端和服务端保持长连接，比如占用30秒（无论30秒是否有数据更新都会建立连接），等30秒的链接时间到后链接取消。
（6）客户端NameServer选择策略:
进行随机取模然后取第一台
4、Broker(Broker充当着消息中转角色，负责存储信息、转发消息。Broker在RocketMq系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求做准备。Broker同时也存储着消息相关的元数据，包括消费组消费进度偏移offset、主题、队列等)。
（1）、Remoting Module:
整个Broker的实体，负责处理来自clients端的请求。而这个Broker实体则油一下模块构成。
（2）、Client Manger:
客户端管理器，负责接收、解析客户端（Producer/Consumer）请求，管理客户端。列如：维护Consumer的Topic订阅信息.
（3）、HA Service：
高可用服务，提供MasterBroker和SlaveBroker之间的数据同步功能。（部署一个Broker之后会产生两个Borker，一个为MasterBroker另一个为SlaveBroker，BrokerName相同但是BrokerId不同，SlaveBroker主要复制MasterBroker的数据）
（4）、IndexService:
索引服务，根据特定的MessageKey，对投递到Broker的信息进行索引服务，同时也提供根据MessageKey对消息进行快速查询的功能
5、同步刷盘和异步刷盘
(1)、同步刷盘增加了数据的可靠性
producer -> java(通过java发送消息) ->memory(内存) -> disk(写入磁盘) -> memory(写入完成)->java (写入完成)
(2)、异步刷盘增加了吞吐量
producer -> java(通过java发送消息) ->memory(内存) -> disk(写入磁盘的同时直接返回写入成功)
6、Broker
（1）、同步复制
当master和slave都成功写入数据之后才返回成功，会降低系统的吞吐量但是能够保持数据的可靠性
(2)、异步复制
当master成功写入数据之后就返回成功，降低了数据的可靠性提高了系统的吞吐量
2、生产者发送消息
发送异步消息
public class AsyncProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
for (int i = 0; i < 10; i++) {
Message mes = new Message("base", "tag1", ("Hello World").getBytes());
producer.send(mes, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
}
@Override
public void onException(Throwable e) {
}
});
TimeUnit.SECONDS.sleep(1L);
}
producer.shutdown();
}
}
发送同步消息
public class SyncProducer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
for (int i = 0; i < 10; i++) {
Message mes = new Message("base", "tag1", ("Hello World").getBytes());
producer.send(mes);
TimeUnit.SECONDS.sleep(1L);
}
producer.shutdown();
}
}
发送单向消息
public class OnewayProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
for (int i = 0; i < 10; i++) {
Message mes = new Message("base", "tag1", ("Hello World").getBytes());
producer.sendOneway(mes);
TimeUnit.SECONDS.sleep(1L);
}
producer.shutdown();
}
}
发送顺序消息
public class Producer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
ArrayList<OrderStep> list = new ArrayList<OrderStep>();
for (int i = 0; i < list.size(); i++) {
String result = list.get(i).toString();
Message mes = new Message("orderTopic", "orderTag", "i" + i, result.getBytes());
producer.send(mes, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Long orderId = (Long) arg;
long index = orderId % mqs.size();
return mqs.get((int)index);
}
},list.get(i).getOrderId());
}
}
}
发送事物消息
public class Producer {
public static void main(String[] args) throws Exception {
TransactionMQProducer producer = new TransactionMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
String tags = msg.getTags();
switch (tags) {
case "TagA":
return LocalTransactionState.COMMIT_MESSAGE;
case "TagB":
return LocalTransactionState.ROLLBACK_MESSAGE;
case "TagC":
return LocalTransactionState.UNKNOW;
}
return LocalTransactionState.UNKNOW;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
return LocalTransactionState.COMMIT_MESSAGE;
}
});
for (int i = 0; i < 10; i++) {
String[] arr = {"TagA", "TagB", "TagC"};
Message mes = new Message("base", arr[i], ("Hello World").getBytes());
producer.sendMessageInTransaction(mes, null);
TimeUnit.SECONDS.sleep(1L);
}
producer.shutdown();
}
}
发送延时消息
public class Producer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
for (int i = 0; i < 10; i++) {
Message mes = new Message("base", "tag1", ("Hello World").getBytes());
mes.setDelayTimeLevel(5);
producer.send(mes);
TimeUnit.SECONDS.sleep(1L);
}
producer.shutdown();
}
}
批量发送消息
public class Producer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
producer.start();
List<Message> msgList = new ArrayList<Message>();
Message mes1 = new Message("base", "tag1", ("Hello World").getBytes());
Message mes2 = new Message("base", "tag2", ("Hello World").getBytes());
Message mes3 = new Message("base", "tag3", ("Hello World").getBytes());
msgList.add(mes1);
msgList.add(mes2);
msgList.add(mes3);
SendResult send = producer.send(msgList);
producer.shutdown();
}
}
3、消费者
普通消费
public class Consumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
consumer.subscribe("base","*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.println(new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
}
}
顺序消费
public class Consumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
consumer.subscribe("orderTopic","*");
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
for (MessageExt msg : msgs) {
System.out.println(msg);
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
}
}
3、整合SpringBoot
# nameserver
rocketmq.name-server=192.168.25.135:9576;192.168.25.138:9876
# 发送者组名
rocketmq.producer.group=my-group
# 消费者组名
rocketmq.consumer.group=my-group
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Test
void testSendMessage() {
rocketMQTemplate.convertAndSend("topic","Hello SpringBoot RocketMq");
}
@RocketMQMessageListener(topic = "topic",consumerGroup = "${rocketmq.consumer.group}",consumeMode = ConsumeMode.CONCURRENTLY )
@Component
public class Consumer implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
}
}
|