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……
阅读全文
2024年3月6日
Nacos 服务注册 nacos-spring-boot-project 中有关服务注册的几个项目 nacos-discovery-spring-boot-actuator nacos-discovery-spring-boot-autoconfigure nacos-discovery-spring-boot-starter 1 2 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ com.alibaba.boot.nacos.discovery.autoconfigure.NacosDiscoveryAutoConfiguration 找到类 NacosDiscoveryAutoConfiguration 1 2 3 4 5 6 7 8 9 10 11 12 13 @ConditionalOnProperty(name = NacosDiscoveryConstants.ENABLED, matchIfMissing = true) @ConditionalOnMissingBean(name = DISCOVERY_GLOBAL_NACOS_PROPERTIES_BEAN_NAME) @EnableNacosDiscovery @EnableConfigurationProperties(value = NacosDiscoveryProperties.class) @ConditionalOnClass(name = "org.springframework.boot.context.properties.bind.Binder") public class NacosDiscoveryAutoConfiguration { @Bean public NacosDiscoveryAutoRegister discoveryAutoRegister() { return new NacosDiscoveryAutoRegister(); } } 注解:EnableNacosDiscovery 1 2 3 4 5 @Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE }) @Retention(RetentionPolicy.RUNTIME) @Documented @Import(NacosDiscoveryBeanDefinitionRegistrar.class) public @interface EnableNacosDiscovery {} import 类 :NacosDiscoveryBeanDefinitionRegistrar 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……
阅读全文
2024年3月6日
一个简单的 Servlet 容器代码设计 Servlet 算是 Java Web 开发请求链路调用栈中底层的一个技术，当客户端发起一个请求后，到达服务器内部，就会先进入 Servlet（这里不讨论更底层的链路），SpringMVC 的请求分发核心也是一个 Servlet，名叫DispatcherServlet，一个请求首先会进入到这……
阅读全文
2024年3月6日
一个简单的 Web 服务器代码设计 在这篇博客中，我们将介绍如何使用 Java 编写一个简单的 Web 服务器。这个 Web 服务器可以接收客户端的 HTTP 请求，并返回一个静态的 HTML 页面。 1. 代码设计 首先，我们需要创建一个 WebServer 类，这个类将负责接收客户端的请求，并返回响应。以下是 WebServer 类的代码： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19……
阅读全文
