RocketMQ启动和消息收发源码分析

前言

简单介绍一下RocketMQ的背景，RocketMQ是阿里开源的消息中间件，根据官网描述，RocketMQ其实是阿里发现ActiveMQ和Kafka无法满足业务需求才开发的。其中很多思想有参考其他消息中间件，具体的几个消息中间件之间的功能特点比较可以参考apache rocketMQ的官方文档，这里就不赘述了。

RocketMQ在官网的整体架构如下：


图片来源于apache官方文档

可以看到，RocketMQ主要包含四大组件：NameServer、Broker、Producer、Consumer。每一个组件都支持水平扩展以预防单点故障。NameServer可以简单理解为注册中心，可用于路由发现和集群管理；Broker则是我们通常意义上的服务端，负责消息的转发、存储、搜索等功能；Producer和Consumer属于客户端，负责消息的发送和接收。

大多数情况下，我们对于RocketMQ的使用停留在服务搭建和简单的消息收发上（如果服务搭建由运维完成的话，这一步也省掉了）。那RocketMQ在NameServer和Broker启动的时候做了什么？Producer又是如何将消息发送到Broker？Consumer的消费流程呢？

本文将基于RocketMQ的4.5.0版本源码，以其中官方提供的example为例，分析一下RocketMQ服务启动和消息发送和接收的流程。

NameServer启动

nameServer的启动入口源码如下，两步走，第一步创建控制器，第二步启动控制器

public static void main(String[] args) {
    main0(args);
}

public static NamesrvController main0(String[] args) {

    try {
        NamesrvController controller = createNamesrvController(args);
        start(controller);
        String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
        log.info(tip);
        System.out.printf("%s%n", tip);
        return controller;
    } catch (Throwable e) {
        e.printStackTrace();
        System.exit(-1);
    }

    return null;
}

创建控制器

看一下创建控制器的源码，有几个点需要注意一下：

  1. 默认会读取ROCKETMQ_HOME系统变量，如果未设置，需要手工指定或设置，否则无法启动
  2. 可以看到NameServer配置来源有三种：默认、配置文件、启动项指定，优先级依次上升（因为后面的配置会覆盖前面的配置）
public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
    // 设置版本号为系统属性
    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    //PackageConflictDetect.detectFastjson();

    // 命令行参数解析的过程
    Options options = ServerUtil.buildCommandlineOptions(new Options());
    commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
    if (null == commandLine) {
        System.exit(-1);
        return null;
    }
    // 初始化NameServer的配置，关键的几个默认配置见$1
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    // 初始化NettyServer的配置，因为RocketMQ各个组件是使用Netty通信的，主要的默认配置见$2
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();
    // NameServer默认的监听端口是9876
    nettyServerConfig.setListenPort(9876);
    // 如果通过 -c 的选项指定了配置配置，会把文件中对应的属性读取到NameServer和NettyServer的配置对象中
    if (commandLine.hasOption('c')) {
        String file = commandLine.getOptionValue('c');
        if (file != null) {
            InputStream in = new BufferedInputStream(new FileInputStream(file));
            properties = new Properties();
            properties.load(in);
            MixAll.properties2Object(properties, namesrvConfig);
            MixAll.properties2Object(properties, nettyServerConfig);

            namesrvConfig.setConfigStorePath(file);

            System.out.printf("load config properties file OK, %s%n", file);
            in.close();
        }
    }
    // 如果指定了 -p 选项，会打印NameServer和NettyServer的配置，然后正常退出
    if (commandLine.hasOption('p')) {
        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
        MixAll.printObjectProperties(console, namesrvConfig);
        MixAll.printObjectProperties(console, nettyServerConfig);
        System.exit(0);
    }

    // 如果启动时指定了某个配置，那放入NameServer的配置对象
    MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

    if (null == namesrvConfig.getRocketmqHome()) {
        System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
        System.exit(-2);
    }
    // 初始化logback的配置
    LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
    JoranConfigurator configurator = new JoranConfigurator();
    configurator.setContext(lc);
    lc.reset();
    configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");

    log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

    MixAll.printObjectProperties(log, namesrvConfig);
    MixAll.printObjectProperties(log, nettyServerConfig);
    // 使用NameServer和NettyServer的配置初始化控制器，同时会初始化控制器中一些其他属性，如KVConfigManager、RouteInfoManager、BrokerHousekeepingService等
    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

    // remember all configs to prevent discard
    // 将所有配置信息注册到一个大的Configuration对象中
    // 其实在NamesrvController构造方法中，NameServer和NettyServer的配置已经注册到Configuration了，而properties包含其中，不知道为什么要再次注册一遍
    controller.getConfiguration().registerConfig(properties);

    return controller;
}

$1 默认的NameServer配置

private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
// kv配置文件路径
private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
// 全局配置文件路径
private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
private String productEnvName = "center";
private boolean clusterTest = false;
// 默认不开启顺序消息
private boolean orderMessageEnable = false;

$2 默认的NettyServer配置

// netty的监听端口(根据代码，会被9876的默认值覆盖)
private int listenPort = 8888;
// Netty默认事件处理线程池，处理如broker注册，topic路由信息查询、topic删除等
private int serverWorkerThreads = 8;
// Netty服务异步回调线程池线程数量
private int serverCallbackExecutorThreads = 0;
// Selector线程数量
private int serverSelectorThreads = 3;
// 控制单向的信号量
private int serverOnewaySemaphoreValue = 256;
// 控制异步信号量
private int serverAsyncSemaphoreValue = 64;
// 服务空闲心跳检测时间间隔 单位秒
private int serverChannelMaxIdleTimeSeconds = 120;
// 发送缓冲区大小
private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
// 接受缓冲区大小
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
// 是否使用Netty内存池
private boolean serverPooledByteBufAllocatorEnable = true;
// 是否启用Epoll IO模型，Linux环境建议开启
private boolean useEpollNativeSelector = false;

接着就是控制器的启动了，start方法如下，==省略了一些基础校验和简单逻辑==，其中最重要的就是controller.initialize()方法

public static NamesrvController start(final NamesrvController controller) throws Exception {
    // controller初始化
    boolean initResult = controller.initialize();
    // controller初始化失败就退出
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }

    // 注册钩子函数，系统正常退出会shutdown掉controller
    Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));
    // 其实就是控制器内部remotingServer的启动，同时异步起一个文件监听服务，注册监听器，详见$5
    controller.start();

    return controller;
}

这里有几个关键点：

  1. 请求处理器有两种，DefaultRequestProcessor和ClusterTestRequestProcessor，由NameServer中的clusterTest字段决定，ClusterTestRequestProcessor继承自DefaultRequestProcessor，唯一的不同就是无法从本地读取路由信息时会从集群中读取
  2. 这里创建了很多的线程池，具体每个线程池的作用见$6
public boolean initialize() {
    // 如果kvConfigPath路径下存在相关文件，会把文件加载到kvConfigManager的配置表中
    this.kvConfigManager.load();
    // 初始化remotingServer，包含了很多nettyServer的相关配置，详见$3
    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    // 创建一个固定大小的线程池，就是Netty中的work线程池，用于处理请求的
    this.remotingExecutor =
        Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    // 这里会注册一个默认的处理器DefaultRequestProcessor，绑定上面创建的线程池，具体能处理哪些请求见$4
    this.registerProcessor();
    // 定时扫描Broker信息并清除失效的Broker路由信息，周期是10秒
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            NamesrvController.this.routeInfoManager.scanNotActiveBroker();
        }
    }, 5, 10, TimeUnit.SECONDS);
    // 定时日志输出kv配置信息，周期10秒
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            NamesrvController.this.kvConfigManager.printAllPeriodically();
        }
    }, 1, 10, TimeUnit.MINUTES);
    // 如果没有关闭TLS，那么会开启一个异步文件监听线程，监听TLS相关配置文件的变化，默认可选的
    if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
        // Register a listener to reload SslContext
        // 略
    }

    return true;
}

$3 NettyRemotingServer构造方法

这里的NettyRemotingServer可以认为就是Netty中的Server，内部其实初始化的基本都是Netty的配置。

public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
    final ChannelEventListener channelEventListener) {
    super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
    this.serverBootstrap = new ServerBootstrap();
    this.nettyServerConfig = nettyServerConfig;
    this.channelEventListener = channelEventListener;
    // 前面的默认配置是0，如果未曾手动设置，这里会设置为4
    int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
    if (publicThreadNums <= 0) {
        publicThreadNums = 4;
    }
    // 新建一个公用的线程池，如果某种消息处理没有注册其他线程则会使用这个
    this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
        // 略
        }
    });
    // 根据是否启用epoll来初始化Netty中的EventLoopGroup
    if (useEpoll()) {
        this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new ThreadFactory() {
            // 略
        });

        this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
            // 略
        });
    } else {
        this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() {
            // 略
        });

        this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
            // 略
        });
    }
    // 加载TLS的上下文信息
    loadSslContext();
}

$4 DefaultRequestProcessor对请求的处理

RemotingCommand有一个code字段用来区分命令类型，可以看到，DefaultRequestProcessor针对不同的RemotingCommand有不同的处理(这里只是一部分命令类型，还有其他命令由其他处理器处理)。常量的名称比较直观，就不一一细说了。==注意，这里用的线程池就是NamesrvController.remotingExecutor==

public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {

    if (ctx != null) {
        log.debug("receive request, {} {} {}",
            request.getCode(),
            RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
            request);
    }

    switch (request.getCode()) {
        case RequestCode.PUT_KV_CONFIG:
            return this.putKVConfig(ctx, request);
        case RequestCode.GET_KV_CONFIG:
            return this.getKVConfig(ctx, request);
        case RequestCode.DELETE_KV_CONFIG:
            return this.deleteKVConfig(ctx, request);
        case RequestCode.QUERY_DATA_VERSION:
            return queryBrokerTopicConfig(ctx, request);
        case RequestCode.REGISTER_BROKER:
            Version brokerVersion = MQVersion.value2Version(request.getVersion());
            if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
                return this.registerBrokerWithFilterServer(ctx, request);
            } else {
                return this.registerBroker(ctx, request);
            }
        case RequestCode.UNREGISTER_BROKER:
            return this.unregisterBroker(ctx, request);
        case RequestCode.GET_ROUTEINTO_BY_TOPIC:
            return this.getRouteInfoByTopic(ctx, request);
        case RequestCode.GET_BROKER_CLUSTER_INFO:
            return this.getBrokerClusterInfo(ctx, request);
        case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
            return this.wipeWritePermOfBroker(ctx, request);
        case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
            return getAllTopicListFromNameserver(ctx, request);
        case RequestCode.DELETE_TOPIC_IN_NAMESRV:
            return deleteTopicInNamesrv(ctx, request);
        case RequestCode.GET_KVLIST_BY_NAMESPACE:
            return this.getKVListByNamespace(ctx, request);
        case RequestCode.GET_TOPICS_BY_CLUSTER:
            return this.getTopicsByCluster(ctx, request);
        case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
            return this.getSystemTopicListFromNs(ctx, request);
        case RequestCode.GET_UNIT_TOPIC_LIST:
            return this.getUnitTopicList(ctx, request);
        case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
            return this.getHasUnitSubTopicList(ctx, request);
        case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
            return this.getHasUnitSubUnUnitTopicList(ctx, request);
        case RequestCode.UPDATE_NAMESRV_CONFIG:
            return this.updateConfig(ctx, request);
        case RequestCode.GET_NAMESRV_CONFIG:
            return this.getConfig(ctx, request);
        default:
            break;
    }
    return null;
}

$5 remotingServer的启动(start方法)

public void start() {
    // 创建线程池处理特定的channel事件(编解码、空闲检测、连接管理等，查看下面具体的Handler)
    this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
        nettyServerConfig.getServerWorkerThreads(),
        new ThreadFactory() {

            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
            }
        });
    // 设置Netty启动类的一些参数
    ServerBootstrap childHandler =
        this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
            .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_KEEPALIVE, false)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
            .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
            .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline()
                        .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
                            new HandshakeHandler(TlsSystemConfig.tlsMode))
                        .addLast(defaultEventExecutorGroup,
                            new NettyEncoder(),
                            new NettyDecoder(),
                            new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                            new NettyConnectManageHandler(),
                            new NettyServerHandler()
                        );
                }
            });
    // 内存池的配置
    if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
        childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    }

    try {
        ChannelFuture sync = this.serverBootstrap.bind().sync();
        InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
        this.port = addr.getPort();
    } catch (InterruptedException e1) {
        throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
    }
    // 这里nettyEventExecutor可以认为是一个线程
    // 会有专门的监听器处理channel事件，这里的监听器就是BrokerHousekeepingService，用来处理Broker的上下线
    // 这里处理的队列任务来源就是NettyConnectManageHandler，NettyConnectManageHandler会针对不同的channel事件将不同任务放入nettyEventExecutor的队列
    if (this.channelEventListener != null) {
        this.nettyEventExecutor.start();
    }

    this.timer.scheduleAtFixedRate(new TimerTask() {

        @Override
        public void run() {
            try {
                // 定时扫描和释放无效的responseFuture
        // responseFuture是一个响应的占位符，如果超时未收到response则清除这个占位符
        // 但是依然会执行响应绑定的回调函数(可能含有超时的处理逻辑)
                NettyRemotingServer.this.scanResponseTable();
            } catch (Throwable e) {
                log.error("scanResponseTable exception", e);
            }
        }
    }, 1000 * 3, 1000);
}

$6 每个线程池的作用(包含Netty线程池)

// 线程数NettyServerConfig.serverWorkerThreads，处理Broker注册、kv配置增删等事件
NamesrvController.remotingExecutor
// 线程数NettyServerConfig.serverCallbackExecutorThreads，从代码来看是Broker处理的时候会用到，如果某个业务处理器未关联线程池就会用这个
NettyRemotingServer.publicExecutor
// 线程数1，Netty中的Boss线程池
NettyRemotingServer.eventLoopGroupBoss
// 线程数NettyServerConfig.serverSelectorThreads，处理channel IO事件
NettyRemotingServer.eventLoopGroupSelector
// 线程数NettyServerConfig.serverWorkerThreads，业务处理线程，很多ChannelHandler绑定的线程池，具体命令处理还会往后提交给其他线程池
NettyRemotingServer.defaultEventExecutorGroup

整个NameServer的启动流程如下：

image

未完待续。。。

Broker启动

消费者启动

生产者启动

生产者发送消息

Broker处理消息

消费者消费消息

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 201,468评论 5赞 473
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 84,620评论 2赞 377
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 148,427评论 0赞 334
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,160评论 1赞 272
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,197评论 5赞 363
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,334评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,775评论 3赞 393
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,444评论 0赞 256
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,628评论 1赞 295
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,459评论 2赞 317
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,508评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,210评论 3赞 318
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,767评论 3赞 303
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,850评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,076评论 1赞 258
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,627评论 2赞 348
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,196评论 2赞 341

推荐阅读更多精彩内容