5分钟学会RocketMQ
2017-07-20 本文已影响45人
H_Man
RocketMQ 简介
-
RocketMQ 是一个队列模型的消息中间件，具有高性能，高可用，高实时等特性，它并不支持JMS（java消息服务）规范，但参考了JMS规范和kafak等的思想。
-
Producer 、Consumer，队列都可以分布式。
-
Producer可以向队列轮流发送消息，队列的集合称为Topic，Consumer可以做广播消费，也可以做集群消费。
-
能够保证严格的消息顺序
-
提供消费者水平订阅扩展能力
-
亿级的消息堆积能力
-
较少的依赖
RocketMQ术语介绍
-
Producer 消息生产者，即生产消息的系统，一般指业务系统。
-
Consumer 消息消费者，即消费消息的系统，一般指后端的异步处理系统。
-
ProducerGroup 生产者组，指一类发送消息的系统。
-
ConsumerGroup 消费组，指消费同一类消息的系统
-
Broker 消息中间件，主要完成消息的存储，转发，是消息系统的核心组件。
-
Topic 消息主题，消息队列的集合。
-
MessageQueue 消息队列，RocketMQ 的所有消息都是持久化的，可以把MessageQueue 理解为一个长度无限的数组，通过下标offset 访问消息
顺序消息
- 消息有序指的是可以按照消息的发送顺序来消费。例如：一笔订单产生了 3 条消息，分别是订单创建、订单付款、订单完成。消费时，要按照顺序依次消费才有意义。与此同时多笔订单之间又是可以并行消费的。首先来看如下示例：
- 假如生产者产生了2条消息：M1、M2，要保证这两条消息的顺序，应该怎样做？你脑中想到的可能是这样
image.png
- 假定M1发送到S1，M2发送到S2，如果要保证M1先于M2被消费，那么需要M1到达消费端被消费后，通知S2，然后S2再将M2发送到消费端。
- 这个模型存在的问题是，如果M1和M2分别发送到两台Server上，就不能保证M1先达到MQ集群，也不能保证M1被先消费。换个角度看，如果M2先于M1达到MQ集群，甚至M2被消费后，M1才达到消费端，这时消息也就乱序了，说明以上模型是不能保证消息的顺序的。如何才能在MQ集群保证消息的顺序？一种简单的方式就是将M1、M2发送到同一个Server上：
image.png
- 这样可以保证M1先于M2到达MQServer（生产者等待M1发送成功后再发送M2），根据先达到先被消费的原则，M1会先于M2被消费，这样就保证了消息的顺序。
- 这个模型也仅仅是理论上可以保证消息的顺序，在实际场景中可能会遇到下面的问题：
image.png
- 只要将消息从一台服务器发往另一台服务器，就会存在网络延迟问题。如上图所示，如果发送M1耗时大于发送M2的耗时，那么M2就仍将被先消费，仍然不能保证消息的顺序。即使M1和M2同时到达消费端，由于不清楚消费端1和消费端2的负载情况，仍然有可能出现M2先于M1被消费的情况。
- 那如何解决这个问题？将M1和M2发往同一个消费者，且发送M1后，需要消费端响应成功后才能发送M2。
- 聪明的你可能已经想到另外的问题：如果M1被发送到消费端后，消费端1没有响应，那是继续发送M2呢，还是重新发送M1？一般为了保证消息一定被消费，肯定会选择重发M1到另外一个消费端2，就如下图所示。
image.png
- 这样的模型就严格保证消息的顺序，细心的你仍然会发现问题，消费端1没有响应Server时有两种情况，一种是M1确实没有到达(数据在网络传送中丢失)，另外一种消费端已经消费M1且已经发送响应消息，只是MQ Server端没有收到。如果是第二种情况，重发M1，就会造成M1被重复消费。
rocketMQ本身没有解决重复消息，如果你的业务中需要禁止重复消息需要自己在业务中去重
image.png
消息消费者
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");
/**
* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
* 如果非第一次启动，那么按照上次消费的位置继续消费
*/
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) {
// 设置自动提交
context.setAutoCommit(true);
for (MessageExt msg : msgs) {
System.out.println(msg + ",内容：" + 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();
}
}
}
