消息队列，常用于应用间通信。
本篇文章基于RocketMQ官方文档
相关概念和组件
概念
Topic：消息分类，依靠topic来定义消息类型
Tag：消息二级分类，可选，同个topic用不同的tag区分消息类别
Message : 泛指MQ所传送的消息体
Producer：消息生产者
Consumer：消息消费者
组件
Name Server：有点类似于zookeeper，负责服务的注册与发现，维护Broker与Topic的映射关系
Broker：负责消息的存储与生产者消费者消息接收与分发，与Name Server建立长连接，保持心跳上传负责的topic信息
Producer：消息生产者，从Name Server获取Broker对应Topic映射关系，然后与Broker建立连接发送消息
Consumer：消息消费者，从Name Server获取Broker对应Topic映射关系，然后与Broker建立连接接收消息
Name Server如果出现宕机，当前Producer和Consumer已经获取的topic路由信息会保持，既有功能不受影响，但若是新增topic和Broker，将会无法传达消息
特点
先进先出
队列的特点，先进先出，不受人为的干预
2.发布-订阅
3.持久化
4.分布式
MQ的定义就是多应用并发通信的高性能中间件。
应用场景
应用解耦
应用间通信不使用MQ会发生并发问题，同时可能会消耗应用资源，影响性能。
通知
应用b一直在监听MQ中来自应用a的消息，逐条消费。
应用分发
MQ是一对多的，可以在多应用间分发（广播）消息。
限制流量
具有队列的特性，这就说明MQ不存在并发问题，同时有且仅有一条数据被消费。
分布式事务
RocketMQ
项目中使用了RocketMQ，这个mq中间件只提供了按topic进行发布-订阅的通信方式，并未提供点对点的消息队列。
涉及的一些概念：
（1）NameServer：在MQ集群中做的是做命名服务，更新和路由发现 broker服务； （2）Broker-Master：broker 消息主机服务器； （3）Broker-Slave：broker 消息从机服务器； （4）Producer：消息生产者； （5）Consumer：消息消费者；
开始使用RocketMQ
quick-start：发送与接收消息
RocketMQ有三种消息发送方式，示例代码均来自于官网。
同步消息，是很可靠的传输方式：
public static void main(String[] args) throws Exception {
//Instantiate with a producer group name.
DefaultMQProducer producer = new
DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
for (int i = 0; i < 100; i++) {
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " +
i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
//Call send message to deliver message to one of brokers.
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
}
//Shut down once the producer instance is not longer in use.
producer.shutdown();
}
异步消息，同样可靠，消息生产者追求效率可以采用此种方式：
public static void main(String[] args) throws Exception {
//Instantiate with a producer group name.
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
producer.setRetryTimesWhenSendAsyncFailed(0);
for (int i = 0; i < 100; i++) {
final int index = i;
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest",
"TagA",
"OrderID188",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("%-10d OK %s %n", index,
sendResult.getMsgId());
}
@Override
public void onException(Throwable e) {
System.out.printf("%-10d Exception %s %n", index, e);
e.printStackTrace();
}
});
}
//Shut down once the producer instance is not longer in use.
producer.shutdown();
}
单向消息，不回传结果，传输可靠性不高，但是效率巨高。比如日志收集：
public static void main(String[] args) throws Exception{
//Instantiate with a producer group name.
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
for (int i = 0; i < 100; i++) {
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " +
i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
//Call send message to deliver message to one of brokers.
producer.sendOneway(msg);
}
//Shut down once the producer instance is not longer in use.
producer.shutdown();
}
接收消息：
public static void main(String[] args) throws InterruptedException, MQClientException {
// Instantiate with specified consumer group name.
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
// Specify name server addresses.
consumer.setNamesrvAddr("localhost:9876");
// Subscribe one more more topics to consume.
consumer.subscribe("TopicTest", "*");
// Register callback to execute on arrival of messages fetched from brokers.
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//Launch the consumer instance.
consumer.start();
System.out.printf("Consumer Started.%n");
}
消息的顺序性保证：Selector
首先需要了解的是，对于部署了多个队列（多个broker）的topic消息，在选择消息队列时采用的是随机算法（或hash、或随机、或依赖时间戳）在消费消息时会轮询Queue列表消费，这样能最大限度的做到负载均衡，但并不能保证消息的顺序性，或者说在同一个topic多节点部署broker时，当消息量很大时，大概率消息的消费顺序是与消息生产不一致的。但有时很多时候消息要保证传输的顺序性，即FIFO，例如支付流程中先后发出的支付请求和支付成功请求。producer可以通过调用
send(Message msg, MessageQueueSelector selector, Object arg)
在发送消息时指定selector，从而按指定的规则选择发送消息的通道（队列）。
MessageQueueSelector是函数式接口：
public interface MessageQueueSelector {
MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}
其中mqs就是消息队列的列表，msg是消息体，arg是自定义的参数。
通过继承该接口定义新的队列选择器，发送消息：
public static void main(String[] args) throws Exception {
//Instantiate with a producer group name.
MQProducer producer = new DefaultMQProducer("example_group_name");
//Launch the instance.
producer.start();
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 100; i++) {
int orderId = i % 10;
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.printf("%s%n", sendResult);
}
//server shutdown
producer.shutdown();
}
同样，消费端也需要做相应的处理
//TODO
参考链接：MQ(消息队列)常见的应用场景解析