RocketMQ知识

摘要：基础概念消息总线，是一种跨进程的通信机制，用于上下游传递消息。每一个都会接收到消息。概念标签，用于对消息分类，在的基础上进行更细的划分。概念中负责接收生产者消息给消费者发送消息的组件。

MQ:
消息总线（Message Queue），是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用MQ之后，消息发送上游只需要依赖MQ，逻辑上和物理上都不用依赖其他服务。
MQ的不足
（1）系统更加复杂，多了一个MQ组件
（2）消息传递路径更长，延时会增加
（3）消息可能会被重复消费
（4）上游无法知道下游的执行结果（因此，调用方实时依赖执行结果的业务场景，请使用调用，而不是MQ）

使用场景
（1）上游不关注执行结果
（2）上游关注结果，但执行时间比较长。举个例子，微信支付，跨公网调用微信的接口，执行时间会比较长，但调用方又非常关注执行结果，此时一般怎么玩呢？

一般采用“回调网关+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 上的消息，方便在开发过程中诊断问题。

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，可以编写一个消费者进行测试

@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 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/manzhizh...
https://www.jianshu.com/p/824...
架构师之路