[关闭]
@vonzhou 2016-09-25T09:04:08.000000Z 字数 44860 阅读 523

RocketMQ - 20160919

MessageQueue RocketMQ

RocketMQ 简介

2016-09-24_095423.png-36.9kB

整体架构

2016-09-24_091833.png-17.2kB

NameServer - 服务发现

轻量名称服务，无状态。

  1. public boolean initialize() {
  2.     /** 从本地文件加载配置信息到内存中 */
  3.     this.kvConfigManager.load();
  5.     /**  启动TCP Server, 使用Netty */
  6.     this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
  8.     this.remotingExecutor =
  9.             Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
  11.     /** 注册TCP消息处理器 */
  12.     this.registerProcessor();
  14.     /** 定期移除非活跃的Broker, broker channel 默认的失效时间是2m */
  15.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  17.         @Override
  18.         public void run() {
  19.             NamesrvController.this.routeInfoManager.scanNotActiveBroker();
  20.         }
  21.     }, 5, 10, TimeUnit.SECONDS);
  23.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  25.         @Override
  26.         public void run() {
  27.             NamesrvController.this.kvConfigManager.printAllPeriodically();
  28.         }
  29.     }, 1, 10, TimeUnit.MINUTES);
  30.     return true;
  31. }

Producer - 发送消息

rocketmq-send.JPG-63kB

　Where？

路由信息来自NameServer。客户端会缓存路由信息TopicPublishInfo, 同时定期从NameServer取Topic路由信息，每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有的NameServer。如果本地查询不到，则需要同步询问NameServer， 如果该Topic不存在，则使用default。

How ?

Producer 在得到了具体的通信地址后，发送过程就显而易见了。通过代码可以看到在选择消息队列进行发送时采用随机方式，同时和上一次发送的broker保持不同，防止热点。

  1. /**
  2.  * 发送消息实现
  3.  */
  4. private SendResult sendDefaultImpl(//
  5.         Message msg,//
  6.         final CommunicationMode communicationMode,//
  7.         final SendCallback sendCallback, final long timeout//
  8. ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
  9.     this.makeSureStateOK();
  10.     Validators.checkMessage(msg, this.defaultMQProducer);
  12.     final long maxTimeout = this.defaultMQProducer.getSendMsgTimeout() + 1000;
  13.     final long beginTimestamp = System.currentTimeMillis();
  14.     long endTimestamp = beginTimestamp;
  15.     /** 获取路由信息 */
  16.     TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
  17.     if (topicPublishInfo != null && topicPublishInfo.ok()) {
  18.         MessageQueue mq = null;
  19.         Exception exception = null;
  20.         SendResult sendResult = null;
  21.         /** 发送失败时的重试时间 */
  22.         int timesTotal = 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed();
  23.         int times = 0;
  24.         String[] brokersSent = new String[timesTotal];
  25.         for (; times < timesTotal && (endTimestamp - beginTimestamp) < maxTimeout; times++) {
  26.             /** 上次发送选择的broker */
  27.             String lastBrokerName = null == mq ? null : mq.getBrokerName();
  28.             /** 选择一个queue */
  29.             MessageQueue tmpmq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
  30.             if (tmpmq != null) {
  31.                 mq = tmpmq;
  32.                 brokersSent[times] = mq.getBrokerName();
  33.                 try {
  34.                     sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
  35.                     endTimestamp = System.currentTimeMillis();
  36.                     switch (communicationMode) {
  37.                     case ASYNC:
  38.                         return null;
  39.                     case ONEWAY:
  40.                         return null;
  41.                     case SYNC:
  42.                         if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
  43.                             if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
  44.                                 continue;
  45.                             }
  46.                         }
  48.                         return sendResult;
  49.                     default:
  50.                         break;
  51.                     }
  52.                 }
  53.                 catch (RemotingException e) {
  54.                     log.warn("sendKernelImpl exception", e);
  55.                     log.warn(msg.toString());
  56.                     exception = e;
  57.                     endTimestamp = System.currentTimeMillis();
  58.                     continue;
  59.                 }
  60.                 catch (MQClientException e) {
  61.                     log.warn("sendKernelImpl exception", e);
  62.                     log.warn(msg.toString());
  63.                     exception = e;
  64.                     endTimestamp = System.currentTimeMillis();
  65.                     continue;
  66.                 }
  67.                 catch (MQBrokerException e) {
  68.                     log.warn("sendKernelImpl exception", e);
  69.                     log.warn(msg.toString());
  70.                     exception = e;
  71.                     endTimestamp = System.currentTimeMillis();
  72.                     switch (e.getResponseCode()) {
  73.                     case ResponseCode.TOPIC_NOT_EXIST:
  74.                     case ResponseCode.SERVICE_NOT_AVAILABLE:
  75.                     case ResponseCode.SYSTEM_ERROR:
  76.                     case ResponseCode.NO_PERMISSION:
  77.                     case ResponseCode.NO_BUYER_ID:
  78.                     case ResponseCode.NOT_IN_CURRENT_UNIT:
  79.                         continue;
  80.                     default:
  81.                         if (sendResult != null) {
  82.                             return sendResult;
  83.                         }
  85.                         throw e;
  86.                     }
  87.                 }
  88.                 catch (InterruptedException e) {
  89.                     log.warn("sendKernelImpl exception", e);
  90.                     log.warn(msg.toString());
  91.                     throw e;
  92.                 }
  93.             }
  94.             else {
  95.                 break;
  96.             }
  97.         } // end of for
  99.         if (sendResult != null) {
  100.             return sendResult;
  101.         }
  103.         String info =
  104.                 String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", //
  105.                     times, //
  106.                     (System.currentTimeMillis() - beginTimestamp), //
  107.                     msg.getTopic(),//
  108.                     Arrays.toString(brokersSent));
  110.         info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
  112.         throw new MQClientException(info, exception);
  113.     }
  115.     List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
  116.     if (null == nsList || nsList.isEmpty()) {
  117.         throw new MQClientException("No name server address, please set it."
  118.                 + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null);
  119.     }
  121.     throw new MQClientException("No route info of this topic, " + msg.getTopic()
  122.             + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO), null);
  123. }
  125. /**
  126.  * 采用 round robin 的方式选择队列，且和上次选择的broker不同
  127.  */
  128. public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
  129.     if (lastBrokerName != null) {
  130.         int index = this.sendWhichQueue.getAndIncrement();
  131.         for (int i = 0; i < this.messageQueueList.size(); i++) {
  132.             int pos = Math.abs(index++) % this.messageQueueList.size();
  133.             MessageQueue mq = this.messageQueueList.get(pos);
  134.             if (!mq.getBrokerName().equals(lastBrokerName)) {
  135.                 return mq;
  136.             }
  137.         }
  139.         return null;
  140.     }
  141.     else {
  142.         int index = this.sendWhichQueue.getAndIncrement();
  143.         int pos = Math.abs(index) % this.messageQueueList.size();
  144.         return this.messageQueueList.get(pos);
  145.     }
  146. }
  148. /**
  149.  * 构造协议消息
  150.  */
  151. private SendResult sendKernelImpl(final Message msg,//
  152.         final MessageQueue mq,//
  153.         final CommunicationMode communicationMode,//
  154.         final SendCallback sendCallback,//
  155.         final long timeout) throws MQClientException, RemotingException, MQBrokerException,
  156.         InterruptedException {
  157.     String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  158.     if (null == brokerAddr) {
  159.         tryToFindTopicPublishInfo(mq.getTopic());
  160.         brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
  161.     }
  163.     SendMessageContext context = null;
  164.     if (brokerAddr != null) {
  165.         if(this.defaultMQProducer.isSendMessageWithVIPChannel()) {
  166.             brokerAddr = MixAll.brokerVIPChannel(brokerAddr);
  167.         }
  169.         byte[] prevBody = msg.getBody();
  170.         try {
  171.             int sysFlag = 0;
  172.             if (this.tryToCompressMessage(msg)) {
  173.                 sysFlag |= MessageSysFlag.CompressedFlag;
  174.             }
  176.             final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  177.             if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
  178.                 sysFlag |= MessageSysFlag.TransactionPreparedType;
  179.             }
  180.             /** 在发送消息前后，预留灵活处理的空间 */
  181.             if (this.hasSendMessageHook()) {
  182.                 context = new SendMessageContext();
  183.                 context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  184.                 context.setCommunicationMode(communicationMode);
  185.                 context.setBornHost(this.defaultMQProducer.getClientIP());
  186.                 context.setBrokerAddr(brokerAddr);
  187.                 context.setMessage(msg);
  188.                 context.setMq(mq);
  189.                 this.executeSendMessageHookBefore(context);
  190.             }
  191.             /** 构造 request header */
  192.             SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
  193.             requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
  194.             requestHeader.setTopic(msg.getTopic());
  195.             requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
  196.             requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
  197.             requestHeader.setQueueId(mq.getQueueId());
  198.             requestHeader.setSysFlag(sysFlag);
  199.             requestHeader.setBornTimestamp(System.currentTimeMillis());
  200.             requestHeader.setFlag(msg.getFlag());
  201.             requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
  202.             requestHeader.setReconsumeTimes(0);
  203.             requestHeader.setUnitMode(this.isUnitMode());
  204.             if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  205.                 String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
  206.                 if (reconsumeTimes != null) {
  207.                     requestHeader.setReconsumeTimes(new Integer(reconsumeTimes));
  208.                     MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
  209.                 }
  210.             }
  212.             SendResult sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(//
  213.                 brokerAddr,// 1
  214.                 mq.getBrokerName(),// 2
  215.                 msg,// 3
  216.                 requestHeader,// 4
  217.                 timeout,// 5
  218.                 communicationMode,// 6
  219.                 sendCallback// 7
  220.                 );
  222.             if (this.hasSendMessageHook()) {
  223.                 context.setSendResult(sendResult);
  224.                 this.executeSendMessageHookAfter(context);
  225.             }
  227.             return sendResult;
  228.         }
  229.         catch (RemotingException e) {
  230.             if (this.hasSendMessageHook()) {
  231.                 context.setException(e);
  232.                 this.executeSendMessageHookAfter(context);
  233.             }
  234.             throw e;
  235.         }
  236.         catch (MQBrokerException e) {
  237.             if (this.hasSendMessageHook()) {
  238.                 context.setException(e);
  239.                 this.executeSendMessageHookAfter(context);
  240.             }
  241.             throw e;
  242.         }
  243.         catch (InterruptedException e) {
  244.             if (this.hasSendMessageHook()) {
  245.                 context.setException(e);
  246.                 this.executeSendMessageHookAfter(context);
  247.             }
  248.             throw e;
  249.         }
  250.         finally {
  251.             msg.setBody(prevBody);
  252.         }
  253.     }
  255.     throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
  256. }
  258. /** 
  259.  * 向特定 broker 发送消息
  260.  * 异步采用 Netty 的 ChannelFutureListener 实现
  261.  */
  262. public SendResult sendMessage(//
  263.         final String addr,// 1
  264.         final String brokerName,// 2
  265.         final Message msg,// 3
  266.         final SendMessageRequestHeader requestHeader,// 4
  267.         final long timeoutMillis,// 5
  268.         final CommunicationMode communicationMode,// 6
  269.         final SendCallback sendCallback// 7
  270. ) throws RemotingException, MQBrokerException, InterruptedException {
  271.     RemotingCommand request = null;
  272.     if (sendSmartMsg) {
  273.         SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
  274.         request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, requestHeaderV2);
  275.     }
  276.     else {
  277.         request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
  278.     }
  280.     request.setBody(msg.getBody());
  282.     switch (communicationMode) {
  283.     case ONEWAY:
  284.         this.remotingClient.invokeOneway(addr, request, timeoutMillis);
  285.         return null;
  286.     case ASYNC:
  287.         this.sendMessageAsync(addr, brokerName, msg, timeoutMillis, request, sendCallback);
  288.         return null;
  289.     case SYNC:
  290.         return this.sendMessageSync(addr, brokerName, msg, timeoutMillis, request);
  291.     default:
  292.         assert false;
  293.         break;
  294.     }
  296.     return null;
  297. }

Broker - 处理来自Producer的消息

每个producer在发送消息的时候都和对应的Broker建立了长连接，此时broker已经准备好接收Message，Broker的SendMessageProcessor.sendMessage处理消息的存储。接收到消息后，会先写入Commit Log文件（顺序写，写满了会新建一个新的文件），然后更新Consume queue文件（存储如何由topic定位到具体的消息）。

捕获2.JPG-72.1kB

  1. /**
  2.  * Broker 处理 Producer发送过来的消息 - SendMessageProcessor
  3.  * 
  4.  */
  5. private RemotingCommand sendMessage(final ChannelHandlerContext ctx, //
  6.         final RemotingCommand request,//
  7.         final SendMessageContext mqtraceContext,//
  8.         final SendMessageRequestHeader requestHeader) throws RemotingCommandException {
  10.     final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
  11.     final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) response.readCustomHeader();
  13.     response.setOpaque(request.getOpaque());
  15.     if (log.isDebugEnabled()) {
  16.         log.debug("receive SendMessage request command, " + request);
  17.     }
  18.     response.setCode(-1);
  19.     super.msgCheck(ctx, requestHeader, response);
  20.     if (response.getCode() != -1) {
  21.         return response;
  22.     }
  24.     final byte[] body = request.getBody();
  26.     /** 消息已经来到了该 broker，只需要知道具体是哪个 queue */
  27.     int queueIdInt = requestHeader.getQueueId();
  28.     TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
  29.     if (queueIdInt < 0) {
  30.         queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
  31.     }
  32.     /** 包装 Message */
  33.     MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
  34.     msgInner.setTopic(requestHeader.getTopic());
  35.     msgInner.setBody(body);
  36.     msgInner.setFlag(requestHeader.getFlag());
  37.     MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
  38.     msgInner.setPropertiesString(requestHeader.getProperties());
  39.     msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(topicConfig.getTopicFilterType(), msgInner.getTags()));
  41.     msgInner.setQueueId(queueIdInt);
  42.     msgInner.setSysFlag(sysFlag);
  43.     msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
  44.     msgInner.setBornHost(ctx.channel().remoteAddress());
  45.     msgInner.setStoreHost(this.getStoreHost());
  46.     msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
  48.     if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
  49.         String traFlag = msgInner.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
  50.         if (traFlag != null) {
  51.             response.setCode(ResponseCode.NO_PERMISSION);
  52.             response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
  53.                     + "] sending transaction message is forbidden");
  54.             return response;
  55.         }
  56.     }
  57.     /** 消息存储 */
  58.     PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
  59.     if (putMessageResult != null) {
  60.         boolean sendOK = false;
  62.         switch (putMessageResult.getPutMessageStatus()) {
  63.         // Success
  64.         case PUT_OK:
  65.             sendOK = true;
  66.             response.setCode(ResponseCode.SUCCESS);
  67.             break;
  68.         case FLUSH_DISK_TIMEOUT:
  69.             response.setCode(ResponseCode.FLUSH_DISK_TIMEOUT);
  70.             sendOK = true;
  71.             break;
  72.         case FLUSH_SLAVE_TIMEOUT:
  73.             response.setCode(ResponseCode.FLUSH_SLAVE_TIMEOUT);
  74.             sendOK = true;
  75.             break;
  76.         case SLAVE_NOT_AVAILABLE:
  77.             response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
  78.             sendOK = true;
  79.             break;
  81.         /** Failed 创建映射文件失败*/
  82.         case CREATE_MAPEDFILE_FAILED:
  83.             response.setCode(ResponseCode.SYSTEM_ERROR);
  84.             response.setRemark("create maped file failed, please make sure OS and JDK both 64bit.");
  85.             break;
  86.         /** .. 其他错误 */
  87.         }
  89.         if (sendOK) {
  90.             this.brokerController.getBrokerStatsManager().incTopicPutNums(msgInner.getTopic());
  91.             this.brokerController.getBrokerStatsManager().incTopicPutSize(msgInner.getTopic(),
  92.                 putMessageResult.getAppendMessageResult().getWroteBytes());
  93.             this.brokerController.getBrokerStatsManager().incBrokerPutNums();
  95.             if (!this.brokerController.getBrokerConfig().isHighSpeedMode()) {
  96.                 // For commercial
  97.                 int incValue =
  98.                         (int) Math.ceil(putMessageResult.getAppendMessageResult().getWroteBytes() / BrokerStatsManager.SIZE_PER_COUNT);
  99.                 this.brokerController.getBrokerStatsManager().incCommercialTopicSendTimes(requestHeader.getProducerGroup(),
  100.                     msgInner.getTopic(), BrokerStatsManager.StatsType.SEND_SUCCESS.toString(), incValue);
  102.                 this.brokerController.getBrokerStatsManager().incCommercialTopicSendSize(requestHeader.getProducerGroup(),
  103.                     msgInner.getTopic(), BrokerStatsManager.StatsType.SEND_SUCCESS.toString(),
  104.                     putMessageResult.getAppendMessageResult().getWroteBytes());
  105.             }
  106.             response.setRemark(null);
  108.             responseHeader.setMsgId(putMessageResult.getAppendMessageResult().getMsgId());
  109.             responseHeader.setQueueId(queueIdInt);
  110.             responseHeader.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());
  111.             /** ACK */
  112.             doResponse(ctx, request, response);
  114.             if (hasSendMessageHook()) {
  115.                 mqtraceContext.setMsgId(responseHeader.getMsgId());
  116.                 mqtraceContext.setQueueId(responseHeader.getQueueId());
  117.                 mqtraceContext.setQueueOffset(responseHeader.getQueueOffset());
  118.             }
  119.             return null;
  120.         }
  121.         else {
  122.             // For commercial
  123.             this.brokerController.getBrokerStatsManager().incCommercialTopicSendTimes(requestHeader.getProducerGroup(),
  124.                 msgInner.getTopic(), BrokerStatsManager.StatsType.SEND_FAILURE.toString(), 1);
  125.         }
  126.     }
  127.     else {
  128.         response.setCode(ResponseCode.SYSTEM_ERROR);
  129.         response.setRemark("store putMessage return null");
  130.     }
  132.     return response;
  133. }
  135. /**
  136.  * DefaultMessageStore 
  137.  */
  138. public PutMessageResult putMessage(MessageExtBrokerInner msg) {
  139.     if (this.shutdown) {
  140.         log.warn("message store has shutdown, so putMessage is forbidden");
  141.         return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
  142.     }
  143.     /** broker slave 不能到这里 */
  144.     if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
  145.         long value = this.printTimes.getAndIncrement();
  146.         if ((value % 50000) == 0) { /** 控制warn log的频率*/
  147.             log.warn("message store is slave mode, so putMessage is forbidden ");
  148.         }
  150.         return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
  151.     }
  152.     /** 不可写 */
  153.     if (!this.runningFlags.isWriteable()) {
  154.         long value = this.printTimes.getAndIncrement();
  155.         if ((value % 50000) == 0) {
  156.             log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits());
  157.         }
  159.         return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
  160.     }
  161.     else {
  162.         this.printTimes.set(0);
  163.     }
  165.     if (msg.getTopic().length() > Byte.MAX_VALUE) {
  166.         log.warn("putMessage message topic length too long " + msg.getTopic().length());
  167.         return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
  168.     }
  170.     if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
  171.         log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
  172.         return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
  173.     }
  175.     long beginTime = this.getSystemClock().now();
  176.     /**
  177.      * 消息写入 commit log
  178.      */
  179.     PutMessageResult result = this.commitLog.putMessage(msg);
  180.     long eclipseTime = this.getSystemClock().now() - beginTime;
  181.     if (eclipseTime > 1000) {
  182.         log.warn("putMessage not in lock eclipse time(ms) " + eclipseTime);
  183.     }
  184.     this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);
  186.     if (null == result || !result.isOk()) {
  187.         this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
  188.     }
  190.     return result;
  191. }
  193. /**
  194.  * CommitLog 写消息
  195.  */
  196. public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
  197.     // Set the storage time
  198.     msg.setStoreTimestamp(System.currentTimeMillis());
  199.     // Set the message body BODY CRC (consider the most appropriate setting
  200.     // on the client)
  201.     msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
  202.     // Back to Results
  203.     AppendMessageResult result = null;
  205.     StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
  207.     String topic = msg.getTopic();
  208.     int queueId = msg.getQueueId();
  210.     final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
  211.     if (tranType == MessageSysFlag.TransactionNotType//
  212.             || tranType == MessageSysFlag.TransactionCommitType) {
  213.         // Delay Delivery
  214.         if (msg.getDelayTimeLevel() > 0) {
  215.             if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
  216.                 msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
  217.             }
  219.             topic = ScheduleMessageService.SCHEDULE_TOPIC;
  220.             queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
  222.             // Backup real topic, queueId
  223.             MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
  224.             MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
  225.             msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
  227.             msg.setTopic(topic);
  228.             msg.setQueueId(queueId);
  229.         }
  230.     }
  232.     long eclipseTimeInLock = 0;
  233.     MapedFile mapedFile = this.mapedFileQueue.getLastMapedFileWithLock();
  234.     synchronized (this) {
  235.         long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
  237.         // Here settings are stored timestamp, in order to ensure an orderly global
  238.         msg.setStoreTimestamp(beginLockTimestamp);
  240.         if (null == mapedFile || mapedFile.isFull()) {
  241.             mapedFile = this.mapedFileQueue.getLastMapedFile();
  242.         }
  244.         if (null == mapedFile) {
  245.             log.error("create maped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
  246.             return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
  247.         }
  248.         /** 写消息到内存映射区 */
  249.         result = mapedFile.appendMessage(msg, this.appendMessageCallback);
  250.         switch (result.getStatus()) {
  251.         case PUT_OK:
  252.             break;
  253.         case END_OF_FILE:
  254.             // Create a new file, re-write the message
  255.             mapedFile = this.mapedFileQueue.getLastMapedFile();
  256.             if (null == mapedFile) {
  257.                 // XXX: warn and notify me
  258.                 log.error("create maped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
  259.                 return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
  260.             }
  261.             result = mapedFile.appendMessage(msg, this.appendMessageCallback);
  262.             break;
  263.         case MESSAGE_SIZE_EXCEEDED:
  264.         case PROPERTIES_SIZE_EXCEEDED:
  265.             return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
  266.         case UNKNOWN_ERROR:
  267.             return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
  268.         default:
  269.             return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
  270.         }
  272.         eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
  273.     } // end of synchronized
  275.     if (eclipseTimeInLock > 500) {
  276.         log.warn("[NOTIFYME]putMessage in lock eclipse time(ms) " + eclipseTimeInLock);
  277.     }
  279.     PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
  281.     // Statistics
  282.     storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
  283.     storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
  285.     GroupCommitRequest request = null;
  287.     /** Synchronization flush 如果是同步刷盘策略 */
  288.     if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
  289.         GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
  290.         if (msg.isWaitStoreMsgOK()) {
  291.             request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
  292.             service.putRequest(request);
  293.             boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
  294.             if (!flushOK) {
  295.                 log.error("do groupcommit, wait for flush failed, topic: " + msg.getTopic() + " tags: " + msg.getTags()
  296.                         + " client address: " + msg.getBornHostString());
  297.                 putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
  298.             }
  299.         } else {
  300.             service.wakeup();
  301.         }
  302.     } else { /** Asynchronous flush 异步刷盘只需要唤醒对应的线程*/
  303.         this.flushCommitLogService.wakeup();
  304.     }
  306.     /** Synchronous write double 如果当前broker是master，则同步双写*/
  307.     if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
  308.         HAService service = this.defaultMessageStore.getHaService();
  309.         if (msg.isWaitStoreMsgOK()) {
  310.             // Determine whether to wait
  311.             if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
  312.                 if (null == request) {
  313.                     request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
  314.                 }
  315.                 service.putRequest(request);
  317.                 service.getWaitNotifyObject().wakeupAll();
  319.                 boolean flushOK =
  320.                 // TODO
  321.                         request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
  322.                 if (!flushOK) {
  323.                     log.error("do sync transfer other node, wait return, but failed, topic: " + msg.getTopic() + " tags: "
  324.                             + msg.getTags() + " client address: " + msg.getBornHostString());
  325.                     putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
  326.                 }
  327.             } else {
  328.                 // Slave problem， Tell the producer, slave not available
  329.                 putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
  330.             }
  331.         }
  332.     }
  334.     return putMessageResult;
  335. }
  337. /**
  338.  * 获得一个映射区，如果不存在则创建
  339.  */
  340. public MapedFile getLastMapedFile(final long startOffset, boolean needCreate) {
  341.     long createOffset = -1;
  342.     MapedFile mapedFileLast = null;
  343.     {
  344.         this.readWriteLock.readLock().lock();
  345.         if (this.mapedFiles.isEmpty()) {
  346.             createOffset = startOffset - (startOffset % this.mapedFileSize);
  347.         }
  348.         else {  /** 获得最近正在使用的 mapped file */
  349.             mapedFileLast = this.mapedFiles.get(this.mapedFiles.size() - 1);
  350.         }
  351.         this.readWriteLock.readLock().unlock();
  352.     }
  354.     if (mapedFileLast != null && mapedFileLast.isFull()) { /** 正在使用的 mapped file 满了 */
  355.         createOffset = mapedFileLast.getFileFromOffset() + this.mapedFileSize;
  356.     }
  358.     if (createOffset != -1 && needCreate) {
  359.         /** 可以看到这里文件命名的规则，以 起始offset作为文件名，便于定位索引 */
  360.         String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
  361.         String nextNextFilePath =
  362.                 this.storePath + File.separator + UtilAll.offset2FileName(createOffset + this.mapedFileSize);
  363.         MapedFile mapedFile = null;
  365.         if (this.allocateMapedFileService != null) {
  366.             mapedFile =
  367.                     this.allocateMapedFileService.putRequestAndReturnMapedFile(nextFilePath,
  368.                         nextNextFilePath, this.mapedFileSize);
  369.         }
  370.         else {
  371.             try {
  372.                 mapedFile = new MapedFile(nextFilePath, this.mapedFileSize);
  373.             }
  374.             catch (IOException e) {
  375.                 log.error("create mapedfile exception", e);
  376.             }
  377.         }
  379.         if (mapedFile != null) {
  380.             this.readWriteLock.writeLock().lock();
  381.             if (this.mapedFiles.isEmpty()) {  /** 标识为该message queue的第一个创建的mapped file */
  382.                 mapedFile.setFirstCreateInQueue(true);
  383.             }
  384.             this.mapedFiles.add(mapedFile); /** 加入链表 */
  385.             this.readWriteLock.writeLock().unlock();
  386.         }
  388.         return mapedFile;
  389.     }
  391.     return mapedFileLast;
  392. }
  394. /**
  395.  * 把消息追加到 mapped file 中，采用回调的方式
  396.  */
  397. public AppendMessageResult appendMessage(final Object msg, final AppendMessageCallback cb) {
  398.     assert msg != null;
  399.     assert cb != null;
  401.     int currentPos = this.wrotePostion.get();
  403.     if (currentPos < this.fileSize) {
  404.         ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
  405.         byteBuffer.position(currentPos);
  406.         AppendMessageResult result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, msg);
  407.         this.wrotePostion.addAndGet(result.getWroteBytes());
  408.         this.storeTimestamp = result.getStoreTimestamp();
  409.         return result;
  410.     }
  412.     log.error("MapedFile.appendMessage return null, wrotePostion: " + currentPos + " fileSize: "
  413.             + this.fileSize);
  414.     return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
  415. }
  417. /**
  418.  * 消息追加实现，同时维护映射关系 (topic-queueId, offset)供消费者使用
  419.  */
  420. public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank, final Object msg) {
  421.     // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
  422.     MessageExtBrokerInner msgInner = (MessageExtBrokerInner) msg;
  423.     // PHY OFFSET
  424.     long wroteOffset = fileFromOffset + byteBuffer.position();
  425.     String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(), wroteOffset);
  427.     // Record ConsumeQueue information
  428.     String key = msgInner.getTopic() + "-" + msgInner.getQueueId();
  429.     Long queueOffset = CommitLog.this.topicQueueTable.get(key);
  430.     if (null == queueOffset) {
  431.         queueOffset = 0L; /** 表示第一条 */
  432.         CommitLog.this.topicQueueTable.put(key, queueOffset);
  433.     }
  435.     /** Transaction messages that require special handling 事务类型的消息 */
  436.     final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
  437.     switch (tranType) {
  438.         // Prepared and Rollback message is not consumed, will not enter the consumer queue
  439.         case MessageSysFlag.TransactionPreparedType:
  440.         case MessageSysFlag.TransactionRollbackType:
  441.             queueOffset = 0L;
  442.             break;
  443.         case MessageSysFlag.TransactionNotType:
  444.         case MessageSysFlag.TransactionCommitType:
  445.         default:
  446.             break;
  447.     }
  449.     /** Serialize message */
  450.     final byte[] propertiesData =
  451.             msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
  452.     if (propertiesData.length > Short.MAX_VALUE) {
  453.         log.warn("putMessage message properties length too long. length={}", propertiesData.length);
  454.         return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
  455.     }
  457.     final short propertiesLength = propertiesData == null ? 0 : (short) propertiesData.length;
  459.     final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
  460.     final int topicLength = topicData == null ? 0 : topicData.length;
  462.     final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
  464.     final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);
  466.     // Exceeds the maximum message
  467.     if (msgLen > this.maxMessageSize) {
  468.         CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
  469.                 + ", maxMessageSize: " + this.maxMessageSize);
  470.         return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
  471.     }
  473.     // Determines whether there is sufficient free space
  474.     if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
  475.         this.resetMsgStoreItemMemory(maxBlank);
  476.         // 1 TOTALSIZE
  477.         this.msgStoreItemMemory.putInt(maxBlank);
  478.         // 2 MAGICCODE
  479.         this.msgStoreItemMemory.putInt(CommitLog.BlankMagicCode);
  480.         // 3 The remaining space may be any value
  481.         // Here the length of the specially set maxBlank
  482.         byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
  483.         return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
  484.             queueOffset);
  485.     }
  487.     /** Initialization of storage space 根据协议格式写入 */
  488.     this.resetMsgStoreItemMemory(msgLen);
  489.     // 1 TOTALSIZE
  490.     this.msgStoreItemMemory.putInt(msgLen);
  491.     // 2 MAGICCODE
  492.     this.msgStoreItemMemory.putInt(CommitLog.MessageMagicCode);
  493.     // 3 BODYCRC
  494.     this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
  495.     // 4 QUEUEID
  496.     this.msgStoreItemMemory.putInt(msgInner.getQueueId());
  497.     // 5 FLAG
  498.     this.msgStoreItemMemory.putInt(msgInner.getFlag());
  499.     // 6 QUEUEOFFSET
  500.     this.msgStoreItemMemory.putLong(queueOffset);
  501.     // 7 PHYSICALOFFSET
  502.     this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
  503.     // 8 SYSFLAG
  504.     this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
  505.     // 9 BORNTIMESTAMP
  506.     this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
  507.     // 10 BORNHOST
  508.     this.msgStoreItemMemory.put(msgInner.getBornHostBytes());
  509.     // 11 STORETIMESTAMP
  510.     this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
  511.     // 12 STOREHOSTADDRESS
  512.     this.msgStoreItemMemory.put(msgInner.getStoreHostBytes());
  513.     // 13 RECONSUMETIMES
  514.     this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
  515.     // 14 Prepared Transaction Offset
  516.     this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
  517.     // 15 BODY
  518.     this.msgStoreItemMemory.putInt(bodyLength);
  519.     if (bodyLength > 0)
  520.         this.msgStoreItemMemory.put(msgInner.getBody());
  521.     // 16 TOPIC
  522.     this.msgStoreItemMemory.put((byte) topicLength);
  523.     this.msgStoreItemMemory.put(topicData);
  524.     // 17 PROPERTIES
  525.     this.msgStoreItemMemory.putShort(propertiesLength);
  526.     if (propertiesLength > 0)
  527.         this.msgStoreItemMemory.put(propertiesData);
  529.     // Write messages to the queue buffer
  530.     byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
  532.     AppendMessageResult result =
  533.             new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId, msgInner.getStoreTimestamp(),queueOffset);
  535.     switch (tranType) {
  536.         case MessageSysFlag.TransactionPreparedType:
  537.         case MessageSysFlag.TransactionRollbackType:
  538.             break;
  539.         case MessageSysFlag.TransactionNotType:  /** 非事务类型 或 事务提交 就更新queue offset，下一个消息存储的位置 */
  540.         case MessageSysFlag.TransactionCommitType:
  541.             // The next update ConsumeQueue information
  542.             CommitLog.this.topicQueueTable.put(key, ++queueOffset);
  543.             break;
  544.         default:
  545.             break;
  546.     }
  548.     return result;
  549. }

Broker - 处理来自Consumer的pull request

2016-09-25_154155.png-50.3kB

  1. /**
  2.  * broker处理消息pull请求
  3.  */
  4. private RemotingCommand processRequest(final Channel channel, RemotingCommand request,
  5.         boolean brokerAllowSuspend) throws RemotingCommandException {
  6.     RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
  7.     final PullMessageResponseHeader responseHeader =
  8.             (PullMessageResponseHeader) response.readCustomHeader();
  9.     final PullMessageRequestHeader requestHeader =
  10.             (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
  12.     response.setOpaque(request.getOpaque());
  14.     SubscriptionGroupConfig subscriptionGroupConfig =
  15.             this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(
  16.                 requestHeader.getConsumerGroup());
  17.     if (null == subscriptionGroupConfig) {
  18.         response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
  19.         response.setRemark("subscription group not exist, " + requestHeader.getConsumerGroup() + " "
  20.                 + FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST));
  21.         return response;
  22.     }
  24.     if (!subscriptionGroupConfig.isConsumeEnable()) {
  25.         response.setCode(ResponseCode.NO_PERMISSION);
  26.         response.setRemark("subscription group no permission, " + requestHeader.getConsumerGroup());
  27.         return response;
  28.     }
  30.     final boolean hasSuspendFlag = PullSysFlag.hasSuspendFlag(requestHeader.getSysFlag());
  31.     final boolean hasCommitOffsetFlag = PullSysFlag.hasCommitOffsetFlag(requestHeader.getSysFlag());
  32.     final boolean hasSubscriptionFlag = PullSysFlag.hasSubscriptionFlag(requestHeader.getSysFlag());
  34.     final long suspendTimeoutMillisLong = hasSuspendFlag ? requestHeader.getSuspendTimeoutMillis() : 0;
  36.     /**  */
  37.     TopicConfig topicConfig =
  38.             this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
  39.     if (null == topicConfig) {
  40.         log.error("the topic " + requestHeader.getTopic() + " not exist, consumer: "
  41.                 + RemotingHelper.parseChannelRemoteAddr(channel));
  42.         response.setCode(ResponseCode.TOPIC_NOT_EXIST);
  43.         response.setRemark("topic[" + requestHeader.getTopic() + "] not exist, apply first please!"
  44.                 + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
  45.         return response;
  46.     }
  48.     if (!PermName.isReadable(topicConfig.getPerm())) {
  49.         response.setCode(ResponseCode.NO_PERMISSION);
  50.         response.setRemark("the topic[" + requestHeader.getTopic() + "] pulling message is forbidden");
  51.         return response;
  52.     }
  54.     if (requestHeader.getQueueId() < 0 || requestHeader.getQueueId() >= topicConfig.getReadQueueNums()) {
  55.         String errorInfo =
  56.                 "queueId[" + requestHeader.getQueueId() + "] is illagal,Topic :"
  57.                         + requestHeader.getTopic() + " topicConfig.readQueueNums: "
  58.                         + topicConfig.getReadQueueNums() + " consumer: " + channel.remoteAddress();
  59.         log.warn(errorInfo);
  60.         response.setCode(ResponseCode.SYSTEM_ERROR);
  61.         response.setRemark(errorInfo);
  62.         return response;
  63.     }
  65.     SubscriptionData subscriptionData = null;
  66.     if (hasSubscriptionFlag) {
  67.         try {
  68.             subscriptionData =
  69.                     FilterAPI.buildSubscriptionData(requestHeader.getConsumerGroup(),
  70.                         requestHeader.getTopic(), requestHeader.getSubscription());
  71.         }
  72.         catch (Exception e) {
  73.             log.warn("parse the consumer's subscription[{}] failed, group: {}",
  74.                 requestHeader.getSubscription(),//
  75.                 requestHeader.getConsumerGroup());
  76.             response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
  77.             response.setRemark("parse the consumer's subscription failed");
  78.             return response;
  79.         }
  80.     }
  81.     else {
  82.         ConsumerGroupInfo consumerGroupInfo =
  83.                 this.brokerController.getConsumerManager().getConsumerGroupInfo(
  84.                     requestHeader.getConsumerGroup());
  85.         if (null == consumerGroupInfo) {
  86.             log.warn("the consumer's group info not exist, group: {}", requestHeader.getConsumerGroup());
  87.             response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
  88.             response.setRemark("the consumer's group info not exist"
  89.                     + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
  90.             return response;
  91.         }
  93.         if (!subscriptionGroupConfig.isConsumeBroadcastEnable() //
  94.                 && consumerGroupInfo.getMessageModel() == MessageModel.BROADCASTING) {
  95.             response.setCode(ResponseCode.NO_PERMISSION);
  96.             response.setRemark("the consumer group[" + requestHeader.getConsumerGroup()
  97.                     + "] can not consume by broadcast way");
  98.             return response;
  99.         }
  101.         subscriptionData = consumerGroupInfo.findSubscriptionData(requestHeader.getTopic());
  102.         if (null == subscriptionData) {
  103.             log.warn("the consumer's subscription not exist, group: {}", requestHeader.getConsumerGroup());
  104.             response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
  105.             response.setRemark("the consumer's subscription not exist"
  106.                     + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
  107.             return response;
  108.         }
  110.         if (subscriptionData.getSubVersion() < requestHeader.getSubVersion()) {
  111.             log.warn("the broker's subscription is not latest, group: {} {}",
  112.                 requestHeader.getConsumerGroup(), subscriptionData.getSubString());
  113.             response.setCode(ResponseCode.SUBSCRIPTION_NOT_LATEST);
  114.             response.setRemark("the consumer's subscription not latest");
  115.             return response;
  116.         }
  117.     }
  119.     final GetMessageResult getMessageResult =
  120.             this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(),
  121.                 requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getQueueOffset(),
  122.                 requestHeader.getMaxMsgNums(), subscriptionData);
  123.     if (getMessageResult != null) {
  124.         response.setRemark(getMessageResult.getStatus().name());
  125.         responseHeader.setNextBeginOffset(getMessageResult.getNextBeginOffset());
  126.         responseHeader.setMinOffset(getMessageResult.getMinOffset());
  127.         responseHeader.setMaxOffset(getMessageResult.getMaxOffset());
  129.         if (getMessageResult.isSuggestPullingFromSlave()) {
  130.             responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig
  131.                 .getWhichBrokerWhenConsumeSlowly());
  132.             log.warn(
  133.                 "consume message too slow, suggest pulling from slave. group={}, topic={}, subString={}, queueId={}, offset={}",
  134.                 requestHeader.getConsumerGroup(), requestHeader.getTopic(),
  135.                 subscriptionData.getSubString(), requestHeader.getQueueId(),
  136.                 requestHeader.getQueueOffset());
  137.         }
  138.         else {
  139.             responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
  140.         }
  142.         switch (getMessageResult.getStatus()) {
  143.         case FOUND:
  144.             response.setCode(ResponseCode.SUCCESS);
  146.             if (this.hasConsumeMessageHook()) {
  147.                 ConsumeMessageContext context = new ConsumeMessageContext();
  148.                 context.setConsumerGroup(requestHeader.getConsumerGroup());
  149.                 context.setTopic(requestHeader.getTopic());
  150.                 context.setClientHost(RemotingHelper.parseChannelRemoteAddr(channel));
  151.                 context.setStoreHost(this.brokerController.getBrokerAddr());
  152.                 context.setQueueId(requestHeader.getQueueId());
  154.                 final SocketAddress storeHost =
  155.                         new InetSocketAddress(brokerController.getBrokerConfig().getBrokerIP1(),
  156.                             brokerController.getNettyServerConfig().getListenPort());
  157.                 Map<String, Long> messageIds =
  158.                         this.brokerController.getMessageStore().getMessageIds(requestHeader.getTopic(),
  159.                             requestHeader.getQueueId(), requestHeader.getQueueOffset(),
  160.                             requestHeader.getQueueOffset() + getMessageResult.getMessageCount(),
  161.                             storeHost);
  162.                 context.setMessageIds(messageIds);
  163.                 context.setBodyLength(getMessageResult.getBufferTotalSize()
  164.                         / getMessageResult.getMessageCount());
  165.                 this.executeConsumeMessageHookBefore(context);
  166.             }
  168.             break;
  169.         case MESSAGE_WAS_REMOVING:
  170.             response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
  171.             break;
  172.         case NO_MATCHED_LOGIC_QUEUE:
  173.         case NO_MESSAGE_IN_QUEUE:
  174.             if (0 != requestHeader.getQueueOffset()) {
  175.                 response.setCode(ResponseCode.PULL_OFFSET_MOVED);
  177.                 log.info(
  178.                     "the broker store no queue data, fix the request offset {} to {}, Topic: {} QueueId: {} Consumer Group: {}",//
  179.                     requestHeader.getQueueOffset(), //
  180.                     getMessageResult.getNextBeginOffset(), //
  181.                     requestHeader.getTopic(),//
  182.                     requestHeader.getQueueId(),//
  183.                     requestHeader.getConsumerGroup()//
  184.                 );
  185.             }
  186.             else {
  187.                 response.setCode(ResponseCode.PULL_NOT_FOUND);
  188.             }
  189.             break;
  190.         case NO_MATCHED_MESSAGE:
  191.             response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
  192.             break;
  193.         case OFFSET_FOUND_NULL:
  194.             response.setCode(ResponseCode.PULL_NOT_FOUND);
  195.             break;
  196.         case OFFSET_OVERFLOW_BADLY:
  197.             response.setCode(ResponseCode.PULL_OFFSET_MOVED);
  198.             log.info("the request offset: " + requestHeader.getQueueOffset()
  199.                     + " over flow badly, broker max offset: " + getMessageResult.getMaxOffset()
  200.                     + ", consumer: " + channel.remoteAddress());
  201.             break;
  202.         case OFFSET_OVERFLOW_ONE:
  203.             response.setCode(ResponseCode.PULL_NOT_FOUND);
  204.             break;
  205.         case OFFSET_TOO_SMALL:
  206.             response.setCode(ResponseCode.PULL_OFFSET_MOVED);
  207.             log.info(
  208.                 "the request offset too small. group={}, topic={}, requestOffset{}, brokerMinOffset={}, clientIp={}",
  209.                 requestHeader.getConsumerGroup(), requestHeader.getTopic(),
  210.                 requestHeader.getQueueOffset(), getMessageResult.getMinOffset(), channel.remoteAddress());
  211.             break;
  212.         default:
  213.             assert false;
  214.             break;
  215.         }
  217.         switch (response.getCode()) {
  218.         case ResponseCode.SUCCESS:
  219.             this.brokerController.getBrokerStatsManager().incGroupGetNums(
  220.                 requestHeader.getConsumerGroup(), requestHeader.getTopic(),
  221.                 getMessageResult.getMessageCount());
  223.             this.brokerController.getBrokerStatsManager().incGroupGetSize(
  224.                 requestHeader.getConsumerGroup(), requestHeader.getTopic(),
  225.                 getMessageResult.getBufferTotalSize());
  227.             this.brokerController.getBrokerStatsManager().incBrokerGetNums(
  228.                 getMessageResult.getMessageCount());
  230.             if(this.brokerController.getBrokerConfig().isTransferMsgByHeap()){
  231.                 final byte[] r = this.readGetMessageResult(getMessageResult);
  232.                 response.setBody(r);
  233.             }
  234.             else {
  235.                 try {
  236.                     FileRegion fileRegion =
  237.                             new ManyMessageTransfer(response.encodeHeader(getMessageResult
  238.                                     .getBufferTotalSize()), getMessageResult);
  239.                     channel.writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
  240.                         @Override
  241.                         public void operationComplete(ChannelFuture future) throws Exception {
  242.                             getMessageResult.release();
  243.                             if (!future.isSuccess()) {
  244.                                 log.error(
  245.                                         "transfer many message by pagecache failed, " + channel.remoteAddress(),
  246.                                         future.cause());
  247.                             }
  248.                         }
  249.                     });
  250.                 } catch (Throwable e) {
  251.                     log.error("transfer many message by pagecache exception", e);
  252.                     getMessageResult.release();
  253.                 }
  255.                 response = null;
  256.             }
  257.             break;
  258.         case ResponseCode.PULL_NOT_FOUND:
  259.             if (brokerAllowSuspend && hasSuspendFlag) {
  260.                 long pollingTimeMills = suspendTimeoutMillisLong;
  261.                 if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
  262.                     pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
  263.                 }
  265.                 PullRequest pullRequest =
  266.                         new PullRequest(request, channel, pollingTimeMills, this.brokerController
  267.                             .getMessageStore().now(), requestHeader.getQueueOffset());
  268.                 this.brokerController.getPullRequestHoldService().suspendPullRequest(
  269.                     requestHeader.getTopic(), requestHeader.getQueueId(), pullRequest);
  270.                 response = null;
  271.                 break;
  272.             }
  274.         case ResponseCode.PULL_RETRY_IMMEDIATELY:
  275.             break;
  276.         case ResponseCode.PULL_OFFSET_MOVED:
  277.             if (this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE
  278.                     || this.brokerController.getMessageStoreConfig().isOffsetCheckInSlave()) {
  279.                 MessageQueue mq = new MessageQueue();
  280.                 mq.setTopic(requestHeader.getTopic());
  281.                 mq.setQueueId(requestHeader.getQueueId());
  282.                 mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());
  284.                 OffsetMovedEvent event = new OffsetMovedEvent();
  285.                 event.setConsumerGroup(requestHeader.getConsumerGroup());
  286.                 event.setMessageQueue(mq);
  287.                 event.setOffsetRequest(requestHeader.getQueueOffset());
  288.                 event.setOffsetNew(getMessageResult.getNextBeginOffset());
  289.                 this.generateOffsetMovedEvent(event);
  290.                 log.warn(
  291.                     "PULL_OFFSET_MOVED:correction offset. topic={}, groupId={}, requestOffset={}, newOffset={}, suggestBrokerId={}",
  292.                     requestHeader.getTopic(), requestHeader.getConsumerGroup(), event.getOffsetRequest(),
  293.                     event.getOffsetNew(), responseHeader.getSuggestWhichBrokerId());
  294.             }
  295.             else {
  296.                 responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
  297.                 response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
  298.                 log.warn(
  299.                     "PULL_OFFSET_MOVED:none correction. topic={}, groupId={}, requestOffset={}, suggestBrokerId={}",
  300.                     requestHeader.getTopic(), requestHeader.getConsumerGroup(),
  301.                     requestHeader.getQueueOffset(), responseHeader.getSuggestWhichBrokerId());
  302.             }
  304.             break;
  305.         default:
  306.             assert false;
  307.         }
  308.     }
  309.     else {
  310.         response.setCode(ResponseCode.SYSTEM_ERROR);
  311.         response.setRemark("store getMessage return null");
  312.     }
  314.     boolean storeOffsetEnable = brokerAllowSuspend;
  315.     storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag;
  316.     storeOffsetEnable = storeOffsetEnable
  317.             && this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;
  318.     if (storeOffsetEnable) {
  319.         this.brokerController.getConsumerOffsetManager().commitOffset(requestHeader.getConsumerGroup(),
  320.             requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
  321.     }
  322.     return response;
  323. }
  325. public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset, final int maxMsgNums,
  326.         final SubscriptionData subscriptionData) {
  327.     if (this.shutdown) {
  328.         log.warn("message store has shutdown, so getMessage is forbidden");
  329.         return null;
  330.     }
  332.     if (!this.runningFlags.isReadable()) {
  333.         log.warn("message store is not readable, so getMessage is forbidden " + this.runningFlags.getFlagBits());
  334.         return null;
  335.     }
  337.     long beginTime = this.getSystemClock().now();
  339.     GetMessageStatus status = GetMessageStatus.NO_MESSAGE_IN_QUEUE;
  340.     long nextBeginOffset = offset;
  341.     long minOffset = 0;
  342.     long maxOffset = 0;
  344.     GetMessageResult getResult = new GetMessageResult();
  346.     final long maxOffsetPy = this.commitLog.getMaxOffset();
  347.     /** consume queue 逻辑消费队列， 也就是消息的索引信息 */
  348.     ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
  349.     if (consumeQueue != null) {
  350.         minOffset = consumeQueue.getMinOffsetInQuque();
  351.         maxOffset = consumeQueue.getMaxOffsetInQuque();
  353.         {
  354.             // 各种临界情况处理
  355.         }
  356.         else { /** 根据拉取消息的offset，得到索引对应的 byte buffer */
  357.             SelectMapedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset);
  358.             if (bufferConsumeQueue != null) {
  359.                 try {
  360.                     status = GetMessageStatus.NO_MATCHED_MESSAGE;
  362.                     long nextPhyFileStartOffset = Long.MIN_VALUE;
  363.                     long maxPhyOffsetPulling = 0;
  365.                     int i = 0;
  366.                     final int MaxFilterMessageCount = 16000;
  367.                     final boolean diskFallRecorded = this.messageStoreConfig.isDiskFallRecorded();
  368.                     for (; i < bufferConsumeQueue.getSize() && i < MaxFilterMessageCount; i += ConsumeQueue.CQStoreUnitSize) {
  369.                         /** 根据 consume queue 的 index 协议格式，读取字段: CommitLogOffset-Size-TagsCode */
  370.                         long offsetPy = bufferConsumeQueue.getByteBuffer().getLong(); // 实际便宜
  371.                         int sizePy = bufferConsumeQueue.getByteBuffer().getInt();// 实际大小
  372.                         long tagsCode = bufferConsumeQueue.getByteBuffer().getLong();
  374.                         maxPhyOffsetPulling = offsetPy;
  376.                         if (nextPhyFileStartOffset != Long.MIN_VALUE) {
  377.                             if (offsetPy < nextPhyFileStartOffset)
  378.                                 continue;
  379.                         }
  381.                         boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy);
  382.                         if (this.isTheBatchFull(sizePy, maxMsgNums, getResult.getBufferTotalSize(), getResult.getMessageCount(),
  383.                             isInDisk)) {
  384.                             break;
  385.                         }
  386.                         /** 如果需要在 broker 进行消息过滤 */
  387.                         if (this.messageFilter.isMessageMatched(subscriptionData, tagsCode)) {
  388.                             /** 具体从 commit log 中读取消息 */
  389.                             SelectMapedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);
  390.                             if (selectResult != null) {
  391.                                 this.storeStatsService.getGetMessageTransferedMsgCount().incrementAndGet();
  392.                                 getResult.addMessage(selectResult);  /** 构造返回的结果 */
  393.                                 status = GetMessageStatus.FOUND;
  394.                                 nextPhyFileStartOffset = Long.MIN_VALUE;
  395.                             }
  396.                             else {
  397.                                 if (getResult.getBufferTotalSize() == 0) {
  398.                                     status = GetMessageStatus.MESSAGE_WAS_REMOVING;
  399.                                 }
  401.                                 nextPhyFileStartOffset = this.commitLog.rollNextFile(offsetPy);
  402.                             }
  403.                         }
  404.                         else {
  405.                             if (getResult.getBufferTotalSize() == 0) {
  406.                                 status = GetMessageStatus.NO_MATCHED_MESSAGE;
  407.                             }
  408.                         }
  409.                     }
  411.                     if (diskFallRecorded) {
  412.                         long fallBehind = maxOffsetPy - maxPhyOffsetPulling;
  413.                         brokerStatsManager.recordDiskFallBehind(group, topic, queueId, fallBehind);
  414.                     }
  416.                     nextBeginOffset = offset + (i / ConsumeQueue.CQStoreUnitSize);
  418.                     long diff = maxOffsetPy - maxPhyOffsetPulling;
  419.                     long memory =
  420.                             (long) (StoreUtil.TotalPhysicalMemorySize * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
  421.                     getResult.setSuggestPullingFromSlave(diff > memory);
  422.                 }
  423.                 finally {
  424.                     bufferConsumeQueue.release();
  425.                 }
  426.             }
  427.             else {
  428.                 status = GetMessageStatus.OFFSET_FOUND_NULL;
  429.                 nextBeginOffset = nextOffsetCorrection(offset, consumeQueue.rollNextFile(offset));
  430.                 log.warn("consumer request topic: " + topic + "offset: " + offset + " minOffset: " + minOffset + " maxOffset: "
  431.                         + maxOffset + ", but access logic queue failed.");
  432.             }
  433.         }
  434.     }
  435.     else {
  436.         status = GetMessageStatus.NO_MATCHED_LOGIC_QUEUE;
  437.         nextBeginOffset = nextOffsetCorrection(offset, 0);
  438.     }
  440.     if (GetMessageStatus.FOUND == status) {
  441.         this.storeStatsService.getGetMessageTimesTotalFound().incrementAndGet();
  442.     }
  443.     else {
  444.         this.storeStatsService.getGetMessageTimesTotalMiss().incrementAndGet();
  445.     }
  446.     long eclipseTime = this.getSystemClock().now() - beginTime;
  447.     this.storeStatsService.setGetMessageEntireTimeMax(eclipseTime);
  449.     getResult.setStatus(status);
  450.     getResult.setNextBeginOffset(nextBeginOffset);
  451.     getResult.setMaxOffset(maxOffset);
  452.     getResult.setMinOffset(minOffset);
  453.     return getResult;
  454. }
  456. /**
  457.  * （topic + queueId) -> ConsumeQueue
  458.  * 在 broker 启动的时候会load所有topic的consume queue信息，运行时不存在则创建
  459.  */
  460. public ConsumeQueue findConsumeQueue(String topic, int queueId) {
  461.     ConcurrentHashMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);
  462.     if (null == map) {
  463.         ConcurrentHashMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);
  464.         ConcurrentHashMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
  465.         if (oldMap != null) { /** 再次确认 */
  466.             map = oldMap;
  467.         }
  468.         else {
  469.             map = newMap;
  470.         }
  471.     } 
  472.     ConsumeQueue logic = map.get(queueId); // 一个逻辑队列
  473.     if (null == logic) {
  474.         ConsumeQueue newLogic = new ConsumeQueue(//
  475.             topic,//
  476.             queueId,//
  477.             StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),//
  478.             this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),//
  479.             this);
  480.         ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
  481.         if (oldLogic != null) {
  482.             logic = oldLogic;
  483.         }
  484.         else {
  485.             logic = newLogic;
  486.         }
  487.     }
  488.     return logic;
  489. }
  491. /**
  492.  * 根据起始索引得到消息索引对应的内存映射区
  493.  */
  494. public SelectMapedBufferResult getIndexBuffer(final long startIndex) {
  495.     int mapedFileSize = this.mapedFileSize;
  496.     long offset = startIndex * CQStoreUnitSize; // 每个索引项的大小是20B
  497.     if (offset >= this.getMinLogicOffset()) {
  498.         MapedFile mapedFile = this.mapedFileQueue.findMapedFileByOffset(offset);
  499.         if (mapedFile != null) {
  500.             SelectMapedBufferResult result = mapedFile.selectMapedBuffer((int) (offset % mapedFileSize));
  501.             return result;
  502.         }
  503.     }
  504.     return null;
  505. }

Consumer - 拉取消息

  1. 根据topic获得路由信息（本地缓存/from nameserver)
  2. 根据路由信息得到 MessageQueue set（相应broker上的所有 read queues）
  3. 构造pull message消息依次轮询各个MessageQueue
  4. 消费消息

Broker - 存储特点

2016-09-25_155207.png-17.8kB

RocketMQ的消息采用顺序写到commitlog文件，然后利用consume queue文件作为索引。RocketMQ采用零拷贝mmap+write的方式来回应Consumer的请求，RocketMQ宣称大部分请求都会在Page Cache层得到满足，所以消息过多不会因为磁盘读使得性能下降，这里自己的理解是，在64bit机器下，虚存地址空间（vm_area_struct）不是问题，所以相关的文件都会被映射到内存中（有定期删除文件的操作），即使此刻不在内存，操作系统也会因为缺页异常进行换入，虽然地址空间不是问题，但是一个进程映射文件的个数(/proc/sys/vm/max_map_count)是有限的，所以可能在这里发生OOM。

rocketmq-broker-index.jpg-25.1kB

Broker存储目录（默认路径是 $HOME/store）：

捕获.JPG-53.2kB

Broker - 刷盘实现

Broker 在消息的存取时直接操作的是内存（内存映射文件），这可以提供系统的吞吐量，但是无法避免机器掉电时数据丢失，所以需要持久化到磁盘中。刷盘的最终实现都是使用NIO中的 MappedByteBuffer.force() 将映射区的数据写入到磁盘，如果是同步刷盘的话，在Broker把消息写到CommitLog映射区后，就会等待写入完成。异步而言，只是唤醒对应的线程，不保证执行的时机。

rocketmq-flush.JPG-37kB

顺序消息是如何保证的？

需要业务层自己决定哪些消息应该顺序到达，然后发送的时候通过规则（hash）映射到同一个队列，因为没有谁比业务自己更加知道关于消息顺序的特点。这样的顺序是相对顺序，局部顺序，因为发送方只保证把这些消息顺序的发送到broker上的同一队列，但是不保证其他Producer也会发送消息到那个队列，所以需要Consumer在拉到消息后做一些过滤。

消息过滤

类似于重复数据删除技术（Data Deduplication），可以在源端做，也可以在目的端实现，就是网络和存储的权衡，如果在Broker端做消息过滤就需要逐一比对consume queue 的 tagsCode 字段（hashcode）,如果符合则传输给消费者，因为是 hashcode，所以存在误判，需要在 Consumer 接收到消息后进行字符串级别的过滤，确保准确性。

小结

参考

RocketMQ Github

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注