您现在的位置是:网站首页 / 大后端大后端

从零开始学RocketMQ(2) - nameServer原理解析

自由达人2021-04-28 1070人围观

简介 nameServer在Rocket中扮演着指挥中心的角色,前面文章提到了Broker启动后会将Broker的信息定时传送给所有指定路径的nameServer,但是nameServer互相是不通讯的,这使得nameServer非常简洁,功能很专一就是维护整个系统的broker路由和集群名,topic等信息。本篇文章分享nameserrver的启动流程,以及如何管理broker的

从零开始学MQ - 你好,RocketMQ》介绍了RocketMQ的使用场景,以及整体架构,每个模块的大体作用和用源码构建了一个实例环境,其中介绍了RocketMQ的大脑组件nameServer,本次就来学习下这个组件的详细实现过程和原理。
首先,额,嗯。上图!

RocketMQ架构图
RocketMQ架构图

可以看到和nameServer有关系的主要有三个,Broker路由,即对Broker的管理(注册与剔除);Broker的发现,这里可以分为两部分一部分是生产端的Broker的发现,还有一部分是消费端的Broker的发现。先整理介绍下nameServer启动流程,从源码角度来分析nameServer内部是如何实现的,入口启动org.apache.rocketmq.namesrv.NamesrvStartup#main
从程序中可以看到createNamesrvController方法是构建控制器的核心方法,
public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
        // 设置版本
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
        //PackageConflictDetect.detectFastjson();

        // 构建命令行参数, 比如 -h输出帮助命令,-n 则是指定nameServer的地址
        Options options = ServerUtil.buildCommandlineOptions(new Options());
        // 构建启动命令行;如果输入的命中中有-h 那就输出刚构建的help信息
        commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
        if (null == commandLine) {
            System.exit(-1);
            return null;
        }

        // nameServer的配置信息类,如ROCKET_HOME,读取默认配置的kvConfigPath,以及存储config的地址等信息
        final NamesrvConfig namesrvConfig = new NamesrvConfig();
        // netty的配置相关信息
        final NettyServerConfig nettyServerConfig = new NettyServerConfig();
        // 指定nameserver端口为9876
        nettyServerConfig.setListenPort(9876);
        // 检查是否指定配置文件-c,如果指定配置文件就读取配置到namesrvConfig和nettyServerConfig
        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参数,就打印参数
        if (commandLine.hasOption('p')) {
            InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
            MixAll.printObjectProperties(console, namesrvConfig);
            MixAll.printObjectProperties(console, nettyServerConfig);
            System.exit(0);
        }

        MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

        // 检查是否配置ROCKET_HOME
        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);
        }

        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);

        final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

        // remember all configs to prevent discard
        // 添加配置
        controller.getConfiguration().registerConfig(properties);

        return controller;
    }

从上面的注释可以看到比较重要的配置有NamesrvConfig,NettyServerConfig。其中NamesrvConfig是配置rocketmqHome,kvConfig配置文件的路径,配置文件储存位置,以及是否开启顺序消息,默认是关闭的。(顺序消息准备在后面发送消息的时候涉及),而NettyServerConfig则是nameserver作为服务器的一些配置,具体如下:


public class NettyServerConfig implements Cloneable {
    // nameserver监听端口,这里定义是8888,其实在后面启动netty的时候写死的9876
    private int listenPort = 8888;
    // netty服务器工作线程数
    private int serverWorkerThreads = 8;
    // public线程数。这是公用线程组,在mq中各自的业务启动各自的线程池,
    // 但如果没有指定线程池的业务就会丢给公用线程池来执行,根据code来匹配。匹配不上就用public线程池
    private int serverCallbackExecutorThreads = 0;
    // netty IO线程数
    private int serverSelectorThreads = 3;
    // 单向消息发送时的信号量,防止压力过大
    private int serverOnewaySemaphoreValue = 256;
    // 异步消息发送时的信号量
    private int serverAsyncSemaphoreValue = 64;
    // netty通道空闲的最长时间 120秒,超过后关系
    private int serverChannelMaxIdleTimeSeconds = 120;
    // netty发送缓冲区大小
    private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
    // netty发送缓冲区大小
    private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
    // byteBuffer是否开启缓存,默认开启
    private boolean serverPooledByteBufAllocatorEnable = true;

    /**
     * make make install
     *
     *
     * ../glibc-2.10.1/configure \ --prefix=/usr \ --with-headers=/usr/include \
     * --host=x86_64-linux-gnu \ --build=x86_64-pc-linux-gnu \ --without-gd
     */
    // 是否启动epoll
    private boolean useEpollNativeSelector = false;
    // 省略get和set方法
}

到此呢 配置项的准备是做完了,接下来是服务器启动

public static NamesrvController start(final NamesrvController controller) throws Exception {

        if (null == controller) {
            throw new IllegalArgumentException("NamesrvController is null");
        }

        // 初始化配置
        boolean initResult = controller.initialize();
        if (!initResult) {
            controller.shutdown();
            System.exit(-3);
        }

        // 注册一个jvm钩子,可以用于优雅停机,在jvm关闭进程的时间先关闭控制器。如果是kill -9 的话当我没说
        Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                controller.shutdown();
                return null;
            }
        }));

        // 启动
        controller.start();

        return controller;
    }

核心方法是initialize();

public boolean initialize() {

        // 是否指定kvConfig的json文件,如果指定了就加载
        this.kvConfigManager.load();
        // 构建服务器,比如netty通讯,初始化公用线程池
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));

        // 注册默认消息处理,用于处理接收到的消息,这里注册的消息处理器其实就用没有匹配到code的请求消息处理器
        this.registerProcessor();

        // 创建一个任务调度线程池,10秒执行一次,检测未活动的broker,RocketMQ定义为两分钟没有心跳的broker即为未活动的
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);

        // 创建一个任务调度线程池。10分钟执行一次,打印所有的配置
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);
        return true;
    }

真正启动netty服务器的是org.apache.rocketmq.remoting.netty.NettyRemotingServer#start

public void start() {
        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连接管理器,netty解码器,serverHandler,shakeHander等
        prepareSharableHandlers();

        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, handshakeHandler)
                            .addLast(defaultEventExecutorGroup,
                                encoder,
                                new NettyDecoder(),
                                new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                connectionManageHandler,
                                serverHandler
                            );
                    }
                });

        if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
            childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }

        // 启动netty服务器
        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);
        }

        if (this.channelEventListener != null) {
            this.nettyEventExecutor.start();
        }

        // 处理作废过期的请求
        this.timer.scheduleAtFixedRate(new TimerTask() {

            @Override
            public void run() {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);
    }

到此为止,nameserver启动的流程是走完了,但是提到的路由管理,Broker发现似乎还不够清晰。那从nameserver单方面的来看看路由是怎么存储的。刚刚从哪么server启动的流程中可以看到nameserver开启了一个定时任务线程池,每隔10秒就会扫描下是否有两分钟没有心跳的Broker,可以看到scanNotActiveBroker()是org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager类中的方法,那么路由信息就存储在RouteInfoManager中,来看看这个路由管理器的内部属性

Broker路由

public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    // 扫描没有心跳的broker的时间间隔,2分钟
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    // 使用到了java的读写锁
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    // 存储topic和队列的关系
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    // 存储Broker的name和对应的Broker信息
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    // 存储Broker集群名和对应实例的名称
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    // 存储broker地址以及该地址下存活的Broker
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    // 存储Broker地址和过滤器的关系
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
}

从RouteInfoManager类可以看出nameserver是用了多个Map来存储topic对应Broker的关系,broker集群,以及存活的broker集合等映射关系,从scanNotActiveBroker()方法可以看出如果最后一次心跳更新时间加上2分钟小于当前时间就删除brokerLiveTable对应的Broker信息并且关闭已经开启的netty连接,并且在onChannelDestroy()方法中进一步的删除信息,如filterServerTable,brokerAddrTable,clusterAddrTable,topicQueueTable集合的缓存信息

public void onChannelDestroy(String remoteAddr, Channel channel) {
        String brokerAddrFound = null;
        if (channel != null) {
            try {
                try {
                    this.lock.readLock().lockInterruptibly();
                    Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
                        this.brokerLiveTable.entrySet().iterator();
                    while (itBrokerLiveTable.hasNext()) {
                        Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
                        if (entry.getValue().getChannel() == channel) {
                            brokerAddrFound = entry.getKey();
                            break;
                        }
                    }
                } finally {
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }

        if (null == brokerAddrFound) {
            brokerAddrFound = remoteAddr;
        } else {
            log.info("the broker's channel destroyed, {}, clean it's data structure at once", brokerAddrFound);
        }

        if (brokerAddrFound != null && brokerAddrFound.length() > 0) {

            try {
                try {
                    this.lock.writeLock().lockInterruptibly();
                    this.brokerLiveTable.remove(brokerAddrFound);
                    this.filterServerTable.remove(brokerAddrFound);
                    String brokerNameFound = null;
                    boolean removeBrokerName = false;
                    Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
                        this.brokerAddrTable.entrySet().iterator();
                    while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
                        BrokerData brokerData = itBrokerAddrTable.next().getValue();

                        Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<Long, String> entry = it.next();
                            Long brokerId = entry.getKey();
                            String brokerAddr = entry.getValue();
                            if (brokerAddr.equals(brokerAddrFound)) {
                                brokerNameFound = brokerData.getBrokerName();
                                it.remove();
                                log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                                    brokerId, brokerAddr);
                                break;
                            }
                        }

                        if (brokerData.getBrokerAddrs().isEmpty()) {
                            removeBrokerName = true;
                            itBrokerAddrTable.remove();
                            log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                                brokerData.getBrokerName());
                        }
                    }

                    if (brokerNameFound != null && removeBrokerName) {
                        Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<String, Set<String>> entry = it.next();
                            String clusterName = entry.getKey();
                            Set<String> brokerNames = entry.getValue();
                            boolean removed = brokerNames.remove(brokerNameFound);
                            if (removed) {
                                log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                                    brokerNameFound, clusterName);

                                if (brokerNames.isEmpty()) {
                                    log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                                        clusterName);
                                    it.remove();
                                }

                                break;
                            }
                        }
                    }

                    if (removeBrokerName) {
                        Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
                            this.topicQueueTable.entrySet().iterator();
                        while (itTopicQueueTable.hasNext()) {
                            Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
                            String topic = entry.getKey();
                            List<QueueData> queueDataList = entry.getValue();

                            Iterator<QueueData> itQueueData = queueDataList.iterator();
                            while (itQueueData.hasNext()) {
                                QueueData queueData = itQueueData.next();
                                if (queueData.getBrokerName().equals(brokerNameFound)) {
                                    itQueueData.remove();
                                    log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                                        topic, queueData);
                                }
                            }

                            if (queueDataList.isEmpty()) {
                                itTopicQueueTable.remove();
                                log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                                    topic);
                            }
                        }
                    }
                } finally {
                    this.lock.writeLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }
    }

从RouteInfoManager类中可以看出nameServe是将broker从注册到心跳检测再到broker超时剔除或者broker关闭后剔除对应的的缓存集合。

Broker发现

从上篇文章可以了解到RocketMQ内部broker是负责消息的接收、存储、转发的,以及topic的创建和删除,从上面的架构图可以看到Broker是支持分布式部署的,生产者、消费者和broker集群可能不在同一个网络环境,那么生产者要发送一个topic为AAA的消息,如何才能知道通过那么Broker来处理呢?就是通过nameServer来查询的,上面提到了一个路由集合topicQueueTable,这个集群存了topic和队列的关系,那么根据topic即可知道该通过哪个Broker来推送消息了。具体的实现可以参照:org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#tryToFindTopicPublishInfo

    private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
        // 先从本地获取缓存,因为producer,consumer都会定期的请求nameServer服务获取信息
        TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
        // 如果获取失败或者为空的时候就再次获取
        if (null == topicPublishInfo || !topicPublishInfo.ok()) {
            this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
        }

        if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
            return topicPublishInfo;
        } else {
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
            return topicPublishInfo;
        }
    }

其中追溯updateTopicRouteInfoFromNameServer(topic)方法到
org.apache.rocketmq.client.impl.MQClientAPIImpl#getTopicRouteInfoFromNameServer(java.lang.String, long, boolean)

public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
        boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
        GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
        requestHeader.setTopic(topic);

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);

        RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.TOPIC_NOT_EXIST: {
                if (allowTopicNotExist) {
                    log.warn("get Topic [{}] RouteInfoFromNameServer is not exist value", topic);
                }

                break;
            }
            case ResponseCode.SUCCESS: {
                byte[] body = response.getBody();
                if (body != null) {
                    return TopicRouteData.decode(body, TopicRouteData.class);
                }
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }

通过项nameServere发送code码为105(GET_ROUTEINFO_BY_TOPIC, 通过topic获取路由信息)的rpc请求来获取路由信息,然后在nameserever服务器上之前提到有注册netty服务器,在org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#processRequest类中可以看到

@Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        if (ctx != null) {
            log.debug("receive request, {} {} {}",
                request.getCode(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                request);
        }
        ...
        case RequestCode.GET_ROUTEINFO_BY_TOPIC:
        return this.getRouteInfoByTopic(ctx, request);
        ...        

可以看到在org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#getRouteInfoByTopic查询信息

// 根据topic查询broker信息
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
// 在pickupTopicRouteData方法中可以看到 根据topic查询brokere信息是从topicQueueTable缓存中获取的,同时构建了生产者所需要的TopicRouteData对象。
public TopicRouteData pickupTopicRouteData(final String topic) {
        TopicRouteData topicRouteData = new TopicRouteData();
        boolean foundQueueData = false;
        boolean foundBrokerData = false;
        Set<String> brokerNameSet = new HashSet<String>();
        List<BrokerData> brokerDataList = new LinkedList<BrokerData>();
        topicRouteData.setBrokerDatas(brokerDataList);

        HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();
        topicRouteData.setFilterServerTable(filterServerMap);

        try {
            try {
                this.lock.readLock().lockInterruptibly();
                List<QueueData> queueDataList = this.topicQueueTable.get(topic);
                if (queueDataList != null) {
                    topicRouteData.setQueueDatas(queueDataList);
                    foundQueueData = true;

                    Iterator<QueueData> it = queueDataList.iterator();
                    while (it.hasNext()) {
                        QueueData qd = it.next();
                        brokerNameSet.add(qd.getBrokerName());
                    }

                    for (String brokerName : brokerNameSet) {
                        BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                        if (null != brokerData) {
                            BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData
                                .getBrokerAddrs().clone());
                            brokerDataList.add(brokerDataClone);
                            foundBrokerData = true;
                            for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
                                List<String> filterServerList = this.filterServerTable.get(brokerAddr);
                                filterServerMap.put(brokerAddr, filterServerList);
                            }
                        }
                    }
                }
            } finally {
                this.lock.readLock().unlock();
            }
        } catch (Exception e) {
            log.error("pickupTopicRouteData Exception", e);
        }

        log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);

        if (foundBrokerData && foundQueueData) {
            return topicRouteData;
        }

        return null;
    }

到此,已经介绍完了nameserver服务器的定位,内部的结构,启动流程项、配置项,以及详细介绍了nameserver的核心作用且分别从源码的角度进行了分析,码字不易请点个赞吧,下篇文章准备分享RocketMQ的消息以及生产者如何发送消息的。

我是小码哥,保持空杯心态,学海无涯,总有你不会的那就一起,欢迎点赞留言。如有错误也请指教

  • 5 点赞
  • 0 收藏
  • 分享

作品评论