Fork me on GitHub

RocketMQ原理

字数统计: 2k字   |   阅读时长≈ 9分

名词

每一个messageQueue都有下面三个东西

brokerOffset

每个broker中的queue在收到消息时会记录offset，初始值为0,每记录一条消息offset会递增+1，

consumerOffset

消费者消费进度/位置

diffTotal

Offset-consumerOffset=diffTotal消费积压/未被消费的消息数量

消费者

DefaultMQPushConsumer 与 DefaultMQPullConsumer

在消费端，我们可以视情况来控制消费过程

DefaultMQPushConsumer 由系统自动控制过程，

DefaultMQPullConsumer 大部分功能需要手动控制

集群消息的消费负载均衡

在集群消费模式下（clustering）

相同的group中的每个消费者只消费topic中的一部分内容

group中的所有消费者都参与消费过程，每个消费者消费的内容不重复，从而达到负载均衡的效果。

使用DefaultMQPushConsumer，新启动的消费者自动参与负载均衡。

ProcessQueue

消息处理类

RocketMQ消息推送原理

consumer与broker采用长轮询方式获取消息

短轮询：

client发送一个请求，server接受之后处理，无论客户端想要的东西有没有，都会立即返回，客户端可能收到空的，也可能收到有数据的，客户端得到服务器反馈之后，会断开连接，

好处：服务器不需要维护客户端的连接

坏处：客户端会发送太多无用的请求

长连接：

client发送连接请求，服务器收到并且维持连接，有数据就推给客户端

好处：数据推送即使

坏处：服务器需要维护连接，开销大

长轮询：

跟短轮询差不多， 差别就在 当服务器没有客户端想要的数据时，会挂起该连接，等到有数据了，在推送数据给客户端

rocketMQ中的broker服务器就是用的这种推送模式，这种模式跟短轮询比肯定是要好一点的，

但是为什么不用长连接，及时推送消息呢，主要原因在于RocketMQ是消息中间件，他不知道客户端的消费能力，推多了 消费端就撑死了，推少了有感觉没有最大利用消费者能力，所以由消费者自己决定他什么时候可以消费

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
DefaultMQPullConsumer  
DefaultMQPushConsumer
//这两个类，他们获取消息的方式，是靠与broker进行长轮询
 //这就是拉取消息的包装类
 this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList)
//这是拉取消息的核心实现方法
 public PullResult pullKernelImpl(
        final MessageQueue mq,
        final String subExpression,
        final String expressionType,
        final long subVersion,
        final long offset,
        final int maxNums,
        final int sysFlag,
        final long commitOffset,
        final long brokerSuspendMaxTimeMillis,
        final long timeoutMillis,
        final CommunicationMode communicationMode,
        final PullCallback pullCallback
    );

源码

MQAdmin

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
MQAdmin
 /*
 MQ管理的基本接口
 
 */
api：

    //创建一个topic ，key是一个访问密钥，queueNum是topic的队列数
    void createTopic(final String key, final String newTopic, final int queueNum)
        throws MQClientException;

   //topicSysFlag 是topic的系统标记
    void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
        throws MQClientException;

    //根据以毫秒为单位的某个时间获取消息队列偏移量<br>由于IO开销较大，请谨慎调用
    long searchOffset(final MessageQueue mq, final long timestamp) throws MQClientException;

    //获取最大偏移量
    long maxOffset(final MessageQueue mq) throws MQClientException;

    //获取最小偏移量
    long minOffset(final MessageQueue mq) throws MQClientException;

    //获取存储的最早消息时间
    long earliestMsgStoreTime(final MessageQueue mq) throws MQClientException;

    //根据消息id查询消息
    MessageExt viewMessage(final String offsetMsgId) throws RemotingException, MQBrokerException,
        InterruptedException, MQClientException;

    //查询消息
    QueryResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
        final long end) throws MQClientException, InterruptedException;

    MessageExt viewMessage(String topic,
        String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException;


//主要作用消息队列的管理

MQConsumer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

 MQConsumer 继承 MQAdmin 
     
    /**
    一个消息队列的获取者
    具备消息成功获取和失败处理的基本能力
     */
     //如果消费失败，调用这个方法将消息将被发送回broker，并延迟消费一段时间，此方法被废弃了（国人框架被废弃的方法，也可能还在框架内部使用，注意）
    @Deprecated
    void sendMessageBack(final MessageExt msg, final int delayLevel) throws RemotingException,
        MQBrokerException, InterruptedException, MQClientException;

    void sendMessageBack(final MessageExt msg, final int delayLevel, final String brokerName)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException;



   //根据topic从comsumer缓存中获取消息队列
    Set<MessageQueue> fetchSubscribeMessageQueues(final String topic) throws MQClientException;



 MQPushConsumer 继承 MQConsumer 
    /**
     * Start the consumer
     */
    void start() throws MQClientException;

    /**
     * Shutdown the consumer
     */
    void shutdown();

    /**
     * Register the message listener
     */
    @Deprecated
    void registerMessageListener(MessageListener messageListener);

    void registerMessageListener(final MessageListenerConcurrently messageListener);

    void registerMessageListener(final MessageListenerOrderly messageListener);

    //订阅主题，subExpression是过滤消息的表达式
    void subscribe(final String topic, final String subExpression) throws MQClientException;

     //通过一个过滤类来过滤，被废弃了
    @Deprecated
    void subscribe(final String topic, final String fullClassName,
        final String filterClassSource) throws MQClientException;

    //使用MessageSelector来过滤
    void subscribe(final String topic, final MessageSelector selector) throws MQClientException;

    //取消订阅
    void unsubscribe(final String topic);

    //动态更新使用者线程池大小
    void updateCorePoolSize(int corePoolSize);

    //暂停消费
    void suspend();

    //恢复消费
    void resume();
}

ClientConfig

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
ClientConfig
/*
客户端的公共配置类

*/
  public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";
    private String namesrvAddr = NameServerAddressUtils.getNameServerAddresses();
    private String clientIP = RemotingUtil.getLocalAddress();
    private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");
    private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
    protected String namespace;
    protected AccessChannel accessChannel = AccessChannel.LOCAL;

    /**
     * Pulling topic information interval from the named server
     */
    private int pollNameServerInterval = 1000 * 30;
    /**
     * Heartbeat interval in microseconds with message broker
     */
    private int heartbeatBrokerInterval = 1000 * 30;
    /**
     * Offset persistent interval for consumer
     */
    private int persistConsumerOffsetInterval = 1000 * 5;
    private long pullTimeDelayMillsWhenException = 1000;
    private boolean unitMode = false;
    private String unitName;
    private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "false"));

    private boolean useTLS = TlsSystemConfig.tlsEnable;

    private LanguageCode language = LanguageCode.JAVA;

客户端消费流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("group2");
        defaultMQPushConsumer.setNamesrvAddr("192.168.47.100:9876");

        MessageSelector selector=MessageSelector.bySql("age=150");
        defaultMQPushConsumer.subscribe("demo",selector);
        defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        //启动
        defaultMQPushConsumer.start();

        System.out.println("consumer  start...");



    }
----上面是一段简单的消费代码-----------
    
   
    
    defaultMQPushConsumer
         consumerGroup：记录此客户端所在的哪个组
         namesrvAddr：记录namesrv服务器地址和端口
         defaultMQPushConsumerImpl=new DefaultMQPushConsumerImpl(this, rpcHook);
              rebalanceImpl=new RebalancePushImpl(this);
                    subscriptionInner：一个容器concurrenthashmap,存放topic和过滤表达式
              messageListenerInner：记录的 消息监听器             
         messageListener：记录的 消息监听器 (我不知道为什么记录两遍)
                        
        
MQClientManager里面的factoryTable存储所有被创建的客户端实例

每一台消费者或者提供者在 broker哪里都是一个MQClientInstance
       MQClientInstance
             nettyClientConfig：new NettyClientConfig();          
             clientConfig：一台客户机克隆下来的配置

                 
  每一台客户机默认接待IO操作的线程数是1
         /*
          this.eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet()));
            }
        });
         
         */
    工作线程：处理请求的命令
        /*
        
        
         this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null);
          public void setClientCallbackExecutorThreads(int clientCallbackExecutorThreads)      {
        this.clientCallbackExecutorThreads = clientCallbackExecutorThreads;
    }
     this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());   
          int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads();
        if (publicThreadNums <= 0) {
            publicThreadNums = 4;
        }

        this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyClientPublicExecutor_" + this.threadIndex.incrementAndGet());
            }
        });

        
        */         
       consumerTable （group，MQConsumerInner)集合  注册该消费者上去
                 
                 
                 
       remotingClient  启动与远程服务器连接
                 
       回调channelHanlder的线程
           /*
           
         defaultMQPushConsumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getNettyClientConfig().setClientWorkerThreads();修改 回调channelhanlder的线程   
           this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
            nettyClientConfig.getClientWorkerThreads(),//默认就是4，没法直接修改
            new ThreadFactory() {

                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());
                }
            });
           
           */
     
                 
           ChannelHandler
                 /*
                   pipeline.addLast(
                        defaultEventExecutorGroup,
                        new NettyEncoder(),
                        new NettyDecoder(),
                        new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
                        new NettyConnectManageHandler(),
                        new NettyClientHandler());
                 
                 */

启动：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
                   if (null == this.clientConfig.getNamesrvAddr()) {
                        this.mQClientAPIImpl.fetchNameServerAddr();
                    }
                    // Start request-response channel
                    this.mQClientAPIImpl.start();
                    // Start various schedule tasks
                    this.startScheduledTask();
                    // Start pull service
                    this.pullMessageService.start();
                    // Start rebalance service
                    this.rebalanceService.start();
                    // Start push service
                    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    



    
    
    pullRequestQueue
      public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                PullRequest pullRequest = this.pullRequestQueue.take();
                this.pullMessage(pullRequest);
            } catch (InterruptedException ignored) {
            } catch (Exception e) {
                log.error("Pull Message Service Run Method exception", e);
            }
        }

        log.info(this.getServiceName() + " service end");
    }
    
     
doRebalance
 public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }

        log.info(this.getServiceName() + " service end");
    }



  public RemotingCommand notifyConsumerIdsChanged(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        try {
            final NotifyConsumerIdsChangedRequestHeader requestHeader =
                (NotifyConsumerIdsChangedRequestHeader) request.decodeCommandCustomHeader(NotifyConsumerIdsChangedRequestHeader.class);
            log.info("receive broker's notification[{}], the consumer group: {} changed, rebalance immediately",
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                requestHeader.getConsumerGroup());
            this.mqClientFactory.rebalanceImmediately();
        } catch (Exception e) {
            log.error("notifyConsumerIdsChanged exception", RemotingHelper.exceptionSimpleDesc(e));
        }
        return null;
    }

一个netty线程连接

4个handler线程
    
4个工作线程，注意处理命令  publicWorkThreads
   
    
一个自定义的监听器处理线程
    
    
一个拉取消息线程
一个rebalace线程
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家