考拉哥的干草堆

rocketmq的顺序消息需要满足2点：

1.Producer端保证发送消息有序，且发送到同一个队列。
2.consumer端保证消费同一个队列。

先看个例子，代码版本跟前面的一样。
Producer类：

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.remoting.exception.RemotingException;


/**
 * Producer，发送顺序消息
 */
public class Producer {
    public static void main(String[] args) throws IOException {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");

            producer.setNamesrvAddr("192.168.0.104:9876");

            producer.start();

            String[] tags = new String[] { "TagA", "TagC", "TagD" };

            Date date = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String dateStr = sdf.format(date);
            for (int i = 0; i < 10; i++) {
                // 加个时间后缀
                String body = dateStr + " Hello RocketMQ " + i;
                Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i, body.getBytes());

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        return mqs.get(id);
                    }
                }, 0);//0是队列的下标

                System.out.println(sendResult + ", body:" + body);
            }

            producer.shutdown();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.in.read();
    }
}

继续阅读 →

源代码版本是3.2.6。接着上一篇消息的产生，这篇是消息的消费。Consumer选择DefaultMQPushConsumer为例。

1.DefaultMQPushConsumer.start()开始。

2.RebalanceService.run()方法定时调用RebalanceImpl.doRebalance()方法，该方法内部是遍历订阅的topic，执行rebalanceByTopic(topic)。

3.调用RebalanceImpl.updateProcessQueueTableInRebalance()，构造PullRequest，从Broker获取nextOffset，pullRequest.setNextOffset(nextOffset)，同时更新本地消费进度记录。

4.调用RebalancePushImpl.dispatchPullRequest(List)。

5.调用PullMessageService.executePullRequestImmediately(final PullRequest)放入pullRequestQueue队列中去。

6.PullMessageService.run()从pullRequestQueue队列中取出PullRequest，调用DefaultMQPushConsumerImpl.pullMessage(pullRequest)作拉取消息的动作。

继续阅读 →

源代码版本是3.2.6。消息的生命周期包括2部分，消息的产生和消息的消费，这篇先说下前者。消息的产生详细一点可以分为:

a.消息产生后由Producer发送至Broker。

b.Broker接收到消息做持久化。

调试代码得到这样的过程，

1.DefaultMQProducer.send()发出消息。

2.DefaultMQProducerImpl.sendDefaultImpl()发出消息。

3.DefaultMQProducerImpl.tryToFindTopicPublishInfo()，即向Namesrv发出GET_ROUTEINTO_BY_TOPIC的请求，来更新
MQProducerInner的topicPublishInfoTable和MQConsumerInner的topicSubscribeInfoTable。

4.调用topicPublishInfo.selectOneMessageQueue()，从发布的topic中轮询取出一个MessageQueue。默认一个topic对应4个MessageQueue。

继续阅读 →

源代码版本是3.2.6。在rocketmq里，consumer被分为2类：MQPullConsumer和MQPushConsumer，其实本质都是拉模式（pull），即consumer轮询从broker拉取消息。

区别是：

push方式里，consumer把轮询过程封装了，并注册MessageListener监听器，取到消息后，唤醒MessageListener的consumeMessage()来消费，对用户而言，感觉消息是被推送过来的。

pull方式里，取消息的过程需要用户自己写，首先通过打算消费的Topic拿到MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，一次取完后，记录该队列下一次要取的开始offset，直到取完了，再换另一个MessageQueue。

继续阅读 →

源代码版本是3.2.6，还是直接跑源代码，启动配置参照前面写的《简介和quickstart》，启动顺序是namesrv，broker，filtersrv，filter和broker有顺序要求，如果filtersrv启动后找不到broker，则会System.exit()退出程序。

看下启动图：

看rocketmq-filtersrv代码，核心processor包下的只有一个Class类且只处理2种类型的请求，即DefaultRequestProcessor.processRequest()只处理RequestCode.REGISTER_MESSAGE_FILTER_CLASS和RequestCode.PULL_MESSAGE：

REGISTER_MESSAGE_FILTER_CLASS：接收consumer端注册过来的filterClass源代码的请求。

PULL_MESSAGE：接收consumer端发出的拉消息的请求。

继续阅读 →

刚刚拉了最新的代码，版本是3.2.6，直接NamesrvStartup类的main()方法启动，不需要带启动参数，启动序列图如下：

当broker，producer，consumer都运行后，namesrv一共有8类线程：

1.ServerHouseKeepingService：守护线程，本质是ChannelEventListener，监听broker的channel变化来更新本地的RouteInfo。

2.NSScheduledThread1：定时任务线程，定时跑2个任务，第一个是，每隔10分钟扫描出不活动的broker，然后从routeInfo中删除，第二个是，每个10分钟定时打印configTable的信息。

3.NettyBossSelector_1:Netty的boss线程（Accept线程），这里只有一根线程。

4.NettyEventExecuter:一个单独的线程，监听NettyChannel状态变化来通知ChannelEventListener做响应的动作。

继续阅读 →