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

推荐阅读更多精彩内容