2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ 消息拉取流程 之前在消费者启动流程中描述过 MQClientInstance 的启动流程，在启动过程中会启动 PullMessageService，它继承了ServiceThread，并且 ServiceThread 实现了 Runnable 接口，所以是单独启动了一个线程 public class PullMessageService extends ServiceThread public abstract class ServiceThread implements Runnable PullMessageService 的 run 方法如下： protected volatile boolean stopped = false; 1……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ broker 处理拉取消息请求流程 org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest(io.netty.channel.ChannelHandlerContext, org.apache.rocketmq.remoting.protocol.RemotingCommand) 第 1 步、校验broker是否可读 1 2 3 4 5 if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) { response.setCode(ResponseCode.NO_PERMISSION); response.setRemark(String.format("the broker[%s] pulling message is forbidden", this.brokerController.getBrokerConfig().getBrokerIP1())); return response; } 第 2 步、根据消费组获取订阅信息 1 2 SubscriptionGroupConfig subscriptionGroupConfig = this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getConsumerGroup()); 第 3 步、校验是否允许消费 1 2 3 4 5 if (!subscriptionGroupConfig.isConsumeEnable()) { response.setCode(ResponseCode.NO_PERMISSION); response.setRemark("subscription group no permission, " + requestHeader.getConsumerGroup()); return response; } 第 4 步、根据主题获取对应的配置信息 1 2 3 4……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ 生产者启动流程 入口： org.apache.rocketmq.client.producer.DefaultMQProducer#start 1 2 3 4 5 6 7 8 9 10 11 12 @Override public void start() throws MQClientException { this.setProducerGroup(withNamespace(this.producerGroup)); this.defaultMQProducerImpl.start(); if (null != traceDispatcher) { try { traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel()); } catch (MQClientException e) { log.warn("trace dispatcher start failed ", e); } } } 第一步、检查 producerGroup 1 2 3 4 5 6 7 8 9 10 11 private void checkConfig() throws MQClientException { Validators.checkGroup(this.defaultMQProducer.getProducerGroup()); if (null == this.defaultMQProducer.getProducerGroup()) { throw new MQClientException("producerGroup is null", null); } if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) { throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",null); } } 第二步、设置 instanceName 1 2……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RockerMQ Nameserver 如何与 Broker 进行通信的？ nameserver 每隔 10s 扫描一次 Broker，移除处于未激活状态的 Broker 核心代码： this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker, 5, 10, TimeUnit.*SECONDS*); 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 public int scanNotActiveBroker() { int removeCount = 0; Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator(); while (it.hasNext()) { Entry<String, BrokerLiveInfo> next = it.next(); long last = next.getValue().getLastUpdateTimestamp(); if ((last +BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) { RemotingUtil.closeChannel(next.getValue().getChannel()); it.remove(); log.warn("The broker channel expired, {} {}ms", next.getKey(),BROKER_CHANNEL_EXPIRED_TIME); this.onChannelDestroy(next.getKey(), next.getValue().getChannel()); removeCount++; } } return removeCount; } broker 每隔 30 秒会向集群……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ MappedFile 内存映射文件详解 1、MappedFile 初始化 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 private void init(final String fileName, final int fileSize) throws IOException { this.fileName = fileName; this.fileSize = fileSize; this.file = new File(fileName); this.fileFromOffset = Long.parseLong(this.file.getName()); boolean ok = false; ensureDirOK(this.file.getParent()); try { this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel(); this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize); TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize); TOTAL_MAPPED_FILES.incrementAndGet(); ok = true; } catch (FileNotFoundException e) { log.error("Failed to create file " + this.fileName, e); throw e; } catch (IOException e) { log.error("Failed to map file……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ IndexFile 详解 首先明确一下 IndexFile 的文件结构 Index header + 哈希槽，每个槽下面挂载 index 索引,类似哈希表的结构 一个 Index 文件默认包含 500 万个哈希槽，一个哈希槽最多存储 4 个 index，也就是一个 IndexFile 默认最多包含 2000 万个 index Index header： 40byte Index header = 8byte 的 beginTimestamp（In……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ 消费者启动流程 org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start 1、检查配置信息 org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#checkConfig 校验消费组的长度不能大于 255 public static final int CHARACTER_MAX_LENGTH = 255; 1 2 3 if (group.length() >CHARACTER_MAX_LENGTH) { throw new MQClientException("the specified group is longer than group max length 255.", null); } 消费组名称只能包含数字、字母、%、-、_、| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 // regex: ^[%|a-zA-Z0-9_-]+$ // % VALID_CHAR_BIT_MAP['%'] = true; // - VALID_CHAR_BIT_MAP['-'] = true; // _ VALID_CHAR_BIT_MAP['_'] = true; // |……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ ConsumeQueue 详解 RocketMQ 基于主题订阅模式实现消息消费，消费者关注每一个主题下的所有消息，但是同一主题下的消息是不连续地存储在 CommitLog 文件中的，如果消费者直接从消息存储文件中遍历查找主题下的消息，效率会特别低。所以为了在查找消息的时候效率更高一些，设计了 ConsumeQueue 文件，可以……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ 消息消费流程 拉取消息 成功之后 会调用 org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService#submitConsumeRequest 组装 消费消息 请求 获取 consumeMessageBatchMaxSize,表示一个 ConsumeRequest 包含的消息 数量，默认为 1 入参 msgs 为拉取消息的最大值，默认为 32 如果 msgs 小于等于 consumeMessageBatchMaxSiz……
阅读全文
2024年3月6日
该文所涉及的 RocketMQ 源码版本为 4.9.3。 RocketMQ CommitLog 详解 commitlog 目录主要存储消息，为了保证性能，顺序写入，每一条消息的长度都不相同，每条消息的前面四个字节存储该条消息的总长度，每个文件大小默认为 1G，文件的命名是以 commitLog 起始偏移量命名的，可以通过修改 broker 配置文件中 mappedFileSizeCommitLog 属性改变文件大小 1、获取最小偏移量 org.apache.rocketmq.store.CommitLog#getMinOffset 1……
阅读全文
