public class FlushRealTimeService extends FlushCommitLogService {
    @Override
    public void run() {
        while (!this.isStopped()) {
            this.waitForRunning(flushInterval);
            commitLog.getStoreCheckpoint().flush(); // 存儲檢查點(diǎn)
            long begin = System.currentTimeMillis();
            boolean result = commitLog.getMappedFileQueue().flush(flushLeastPages);
            logFlushResult(result, begin);
        }
    }
}

RocketRemotingExecutor#processRequest
public void processRequest(ChannelHandlerContext ctx, RemotingCommand request) {
    final int opaque = request.getOpaque();
    final RequestTask task = new RequestTask(ctx, request, opaque);
    executor.submit(task);
}

brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
flushDiskType=ASYNC_FLUSH
flushCommitLogLeastPages=4
brokerSuspendMaxTimeMillis=2000
brokerCommitLogRetainTime=72
storePathRootDir=/data/rocketmq/store
storePathCommitLog=/data/rocketmq/store/commitlog
storePathConsumeQueue=/data/rocketmq/store/consumequeue
storePathIndex=/data/rocketmq/store/index
messageIndexEnable=true
brokerCallbackExecutorThreads=8
sendMessageThreadPoolNums=16
pullMessageThreadPoolNums=16

public class ProducerExample {
  public static void main(String[] args) throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("PID_SECKILL_GROUP");
    producer.setNamesrvAddr("nameserver1:9876;nameserver2:9876");
    producer.setSendMsgTimeout(3000);
    producer.setRetryTimesWhenSendFailed(2);
    // 啟用批量發(fā)送
    producer.setMaxMessageSize(4 * 1024 * 1024);
    producer.start();

    for (int i = 0; i < 1000000; i++) {
      Message msg = new Message(
        "Topic_Seckill",
        "TagA",
        ("秒殺請求-" + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
      );
      SendResult result = producer.send(msg, new MessageQueueSelector() {
        @Override
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
          int id = ((Long)arg).intValue();
          return mqs.get(id % mqs.size());
        }
      }, ThreadLocalRandom.current().nextInt());
      if (i % 10000 == 0) {
        System.out.printf("Send %d msgs, result=%s%n", i, result.getSendStatus());
      }
    }
    producer.shutdown();
  }
}

public class ConsumerExample {
  public static void main(String[] args) throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_SECKILL_GROUP");
    consumer.setNamesrvAddr("nameserver1:9876;nameserver2:9876");
    consumer.setConsumeThreadMin(20);
    consumer.setConsumeThreadMax(64);
    consumer.subscribe("Topic_Seckill", "TagA||TagB");

    consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
      for (MessageExt msg : msgs) {
        // 業(yè)務(wù)處理邏輯
        System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8));
      }
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
    System.out.printf("Consumer Started.%n");
  }
}