RocketMQ知识（及开发实战）
MQ基础概念：
-
MQ:
消息总线（Message Queue），是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用MQ之后，消息发送上游只需要依赖MQ，逻辑上和物理上都不用依赖其他服务。
MQ的不足
（1）系统更加复杂，多了一个MQ组件
（2）消息传递路径更长，延时会增加
（3）消息可能会被重复消费
（4）上游无法知道下游的执行结果（因此，调用方实时依赖执行结果的业务场景，请使用调用，而不是MQ）使用场景
（1）上游不关注执行结果
（2）上游关注结果，但执行时间比较长。举个例子，微信支付，跨公网调用微信的接口，执行时间会比较长，但调用方又非常关注执行结果，此时一般怎么玩呢？
image.png
一般采用“回调网关+MQ”方案来解耦：
a、调用方直接跨公网调用微信接口
b、微信返回调用成功，此时并不代表返回成功
c、微信执行完成后，回调统一网关
d、网关将返回结果通知MQ
e、请求方收到结果通知
-
rocketMQ：
RocketMQ 是什么？
Github 上关于 RocketMQ 的介绍：
RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
支持拉（pull）和推（push）两种消息模式
单一队列百万消息的堆积能力
支持多种消息协议，如 JMS、MQTT 等
分布式高可用的部署架构,满足至少一次消息传递语义
提供 docker 镜像用于隔离测试和云集群部署
提供配置、指标和监控等功能丰富的 Dashboard -
consumer group：
1、概念：消费者分组，多个消费者在一个消费者分组中。
2、注意点：一个consumer group中的机器相当于一个集群，consumer group中只有一台机器会接收到消息，并进行消费。每一个consumer group都会接收到消息。这样子的设计要求消费端需要保证幂等性。 -
topic：
1、概念：Topic 是一种消息的逻辑分类，比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类，一个是订单 Topic 存放订单相关的消息，一个是库存 Topic 存储库存相关的消息。
2、生产方发出的消息绑定某个topic，然后消费方监听某个topic，消费方（各个group）接收到消息，进行消费
3、topic应用级别：整个应用最好都使用一个topic，而更加细的区分，使用tags来区分。 -
tag:
1、概念：标签，用于对消息分类，在topic的基础上进行更细的划分。 -
nameServer：
1、概念：Name Server 为 producer 和 consumer 提供路由信息。类似rpc中的注册中心。当producer需要发送消息首先去询问nameServer需要请求哪个broker。而当consumer需要拉取消息，也会先询问nameServer需要请求哪个broker。 -
broker:
1、概念：rocketMQ中负责接收生产者消息、给消费者发送消息的组件。 -
Message：
1、概念：Message 是消息的载体。一个 Message 必须指定 topic。Message 还有一个可选的 tag 设置，以便消费端可以基于 tag 进行过滤消息。也可以添加额外的键值对，例如你需要一个业务 key 来查找 broker 上的消息，方便在开发过程中诊断问题。
MQ生产者者实例：
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
//声明并初始化一个producer
//需要一个producer group名字作为构造方法的参数，这里为producer1
DefaultMQProducer producer = new DefaultMQProducer("producer1");
//设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
//调用start()方法启动一个producer实例
producer.start();
//发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
for (int i = 0; i < 10; i++) {
try {
Message msg = new Message("TopicTest",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
);
//调用producer的send()方法发送消息
//这里调用的是同步的方式，所以会有返回结果
SendResult sendResult = producer.send(msg);
//打印返回结果，可以看到消息发送的状态以及一些相关信息
System.out.println(sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
//发送完消息之后，调用shutdown()方法关闭producer
producer.shutdown();
}
}
MQ消费者实例：
在开发过程中，如果想测试生产者是否发出了mq，可以编写一个消费者进行测试
@Test
public void testMqConsumer() throws Exception {
String rocketmqAddress="10.113.41.2:9876;10.113.41.4:9876";
int threadNum = 5;
String topics = "WechatUnionCoreTemplateNotifyTopic";
String instanceName = "TemplateComsumer";
String groupName = "wechatUnionTemplateNotifyConsumer";
DefaultMQPushConsumer consumer = null;
consumer = new DefaultMQPushConsumer(groupName);
consumer.setNamesrvAddr(rocketmqAddress);//MQ地址
consumer.setClientCallbackExecutorThreads(threadNum);//消费现场数量
consumer.setInstanceName(instanceName);//实例名称
consumer.subscribe(topics, "*");
//注册监听
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (int i = 0; i < msgs.size(); i++) {
MessageExt msgExt = msgs.get(i);
String msgId = msgExt.getMsgId();
Integer flag = msgExt.getFlag();
TemplateNotifyItem templateNotifyItem = ProtoBufSerialize.fromProto(msgExt.getBody(), TemplateNotifyItem.class);
logger.info("receive new Msg: " + " msgId=" + msgId + " flag=" + flag + " templateNotifyItem=" + templateNotifyItem);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
logger.info("监听执行中");
Thread.sleep(1000000);
}
参考：
http://blog.csdn.net/manzhizhen/article/details/52606733
https://www.jianshu.com/p/824066d70da8
架构师之路-mq系列
