RocketMQ——RocketMQ消息存储

作者: fffhJk | 来源:发表于2019-12-26 19:12 被阅读0次

DefaultMQPushConsumer

属性
consumerGroup	消费组名称
messageModel	消息消费模式，分为集群模式和广播模式
consumeFromWhere	消费者开始消费的位置，默认为最大偏移量 CONSUME_FROM_LAST_OFFSET
allocateMessageQueueStrategy	集群模式下消费队列负载均衡策略
subscription	订阅信息
messageListener	消息业务监听器
offsetStore	消费进度存储器
consumeThreadMin	消费者最小线程数
consumeThreadMax	消费最大线程数
pullBatchSize	每次拉取消息size
DefaultMQPushConsumerImpl	核心实现，核心的方法都在这里实现

消费者启动流程

DefaultMQPushConsumer#start

 this.defaultMQPushConsumerImpl.start();

DefaultMQPushConsumerImpl#start

创建MQClient

this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);

负载均衡初始化

this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);

初始化PullAPIWrapper

 this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

初始化 offsetStore 集群模式文件是存储在broker，而广播模式文件是存储在本地。

 switch (this.defaultMQPushConsumer.getMessageModel()) {
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }

consumeMessageService.start()

    public void start() {
        this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                cleanExpireMsg();
            }

        }, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);
    }

注册consumer

 boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);

MQClient启动

 mQClientFactory.start();

   public void start() throws MQClientException {

        synchronized (this) {
            switch (this.serviceState) {
                case CREATE_JUST:
                    this.serviceState = ServiceState.START_FAILED;
                    // If not specified,looking address from name server
                    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);
                    log.info("the client factory [{}] start OK", this.clientId);
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                    break;
                case SHUTDOWN_ALREADY:
                    break;
                case START_FAILED:
                    throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
                default:
                    break;
            }
        }
    }

收尾

 this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        this.mQClientFactory.checkClientInBroker();
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        this.mQClientFactory.rebalanceImmediately();

消息拉取

看到再MQ consumer启动过程中，会启动 mQClientFactory.start() 方法中，会启动对应的this.pullMessageService.start()
我们查看对应的PullMessageService。

PullMessageService

    @Override
    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");
    }

PullRequest 的添加如下：
PullMessageService#executePullRequestLater

    public void executePullRequestLater(final PullRequest pullRequest, final long timeDelay) {
        if (!isStopped()) {
            this.scheduledExecutorService.schedule(new Runnable() {
                @Override
                public void run() {
                    PullMessageService.this.executePullRequestImmediately(pullRequest);
                }
            }, timeDelay, TimeUnit.MILLISECONDS);
        } else {
            log.warn("PullMessageServiceScheduledThread has shutdown");
        }
    }

    public void executePullRequestImmediately(final PullRequest pullRequest) {
        try {
            this.pullRequestQueue.put(pullRequest);
        } catch (InterruptedException e) {
            log.error("executePullRequestImmediately pullRequestQueue.put", e);
        }
    }

DefaultMQPushConsumerImpl#pullMessage 则调用对应的方法。

PullRequest 简介

PullRequest结构如下

属性
private String consumerGroup	消费者组
private MessageQueue messageQueue	待拉取消费队列
private ProcessQueue processQueue	消息处理队列，从broker拉取的消息先存储到ProcessQueue，然后再提交到消费者线程池消费
private long nextOffset	待拉取的MessageQueue偏移量
private boolean lockedFirst = false	是否被锁定

MessageQueue结构如下：

属性
private String topic	topic
private String brokerName	brokername
private int queueId	queueId

PullMessageService#pullMessage

    private void pullMessage(final PullRequest pullRequest) {
        final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
        if (consumer != null) {
            DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
            impl.pullMessage(pullRequest);
        } else {
            log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
        }
    }

ProcessQueue

ProcessQueue 用来存储broker拉取的消息，是MessageQueue在消费端的重现和快照。 PullMessageService每次默认拉取32条消息，按消息的队列偏移量顺序存在ProcessQueue中，PullMessageService将消息提交到消费者线程池，消费成功后从ProcessQueue中移除。
属性简介


ReadWriteLock lockTreeMap = new ReentrantReadWriteLock()
TreeMap<Long, MessageExt> msgTreeMap = new TreeMap<Long, MessageExt>()	消息存储器，key为消息再ConsumeQueue中的偏移量，MessageExt为消息体
AtomicLong msgCount = new AtomicLong()	ProcessQueue中的消息总量
AtomicLong msgSize = new AtomicLong()
Lock lockConsume = new ReentrantLock()
TreeMap<Long, MessageExt> consumingMsgOrderlyTreeMap	顺序消息消费
AtomicLong tryUnlockTimes = new AtomicLong(0)
volatile long queueOffsetMax = 0L	当前ProcessQueue中包含的最大队列偏移量
volatile boolean dropped = false	当前ProcessQueue是否被丢弃
volatile long lastPullTimestamp = System.currentTimeMillis()	上次拉取消息时间戳
volatile long lastConsumeTimestamp = System.currentTimeMillis()	上次消息消费时间戳
volatile boolean locked = false
volatile long lastLockTimestamp = System.currentTimeMillis()	上次锁定时间戳
volatile boolean consuming = false
volatile long msgAccCnt = 0

消息拉取流程

消息拉取分为三个流程：

消息拉取客户端消息拉取请求封装
消费服务器查找并返回消息
消息拉取客户端处理返回的消息

如下图为consumer启动然后拉取消息的流程

D479A36DA39B4460DEC046151C77B517.jpg

最终PullMessageService 调用DefaultMQPushConsumerImpl中的pullMessage
DefaultMQPushConsumerImpl#pullMessage 核心调用方法：

 try {
            this.pullAPIWrapper.pullKernelImpl(
                pullRequest.getMessageQueue(),//拉取的消息队列
                subExpression,//消息过滤表达式
                subscriptionData.getExpressionType(),//消息表达式类型，TAG、SQL92
                subscriptionData.getSubVersion(),
                pullRequest.getNextOffset(),//消息拉取偏移量
                this.defaultMQPushConsumer.getPullBatchSize(),//本次拉取最大消息条数，默认32条
                sysFlag,//拉取系统标记
                commitOffsetValue,//当前MessageQueue的消费进度
                BROKER_SUSPEND_MAX_TIME_MILLIS,//消息拉取过程中允许broker挂起时间
                CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,//消息拉取超时时间
                CommunicationMode.ASYNC,//消息拉取模式，默认为异步拉取
                pullCallback//从broker拉取到消息后的回调方法
            );
        } catch (Exception e) {
            log.error("pullKernelImpl exception", e);
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
        }

PullAPIWrapper#pullKernelImpl

获取broker信息
封装request
发起调用，执行回调函数

MQClientAPIImpl#pullMessageSync

 this.remotingClient.invokeSync(addr, request, timeoutMillis)

Broker组装消息

RequestCode#PULL_MESSAGE 定位到Broker端处理消息拉取的入口 PullMessageProcessor#processRequest
核心代码只有一行

 final GetMessageResult getMessageResult =
            this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
                requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);

MQClientAPIImpl#processPullResponse

在 PullAPIWrapper#pullKernelImpl 方法执行完毕后，执行 processPullResponse 来处理返回response

pullCallback.onSuccess(pullResult);

执行pullCallback回调函数

消息长轮询模式

RocketMQ并没有真正实现Push模式，而是循环向服务端发送拉取消息请求，拉取消息。

消息队列负载和重新分布机制

RocketMQ的负载均衡通过RebalanceService实现，每个MQClientInstance都持有RebalanceService实例。
RebalanceService#run


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

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

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

RebalanceImpl#doRebalance

    public void doRebalance(final boolean isOrder) {
        Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
        if (subTable != null) {
            for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
                final String topic = entry.getKey();
                try {
                    this.rebalanceByTopic(topic, isOrder);
                } catch (Throwable e) {
                    if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        log.warn("rebalanceByTopic Exception", e);
                    }
                }
            }
        }

        this.truncateMessageQueueNotMyTopic();
    }