关于mq:mq从零开始实现-mq04启动检测与实现优化

前景回顾

【mq】从零开始实现 mq-01- 生产者、消费者启动

【mq】从零开始实现 mq-02- 如何实现生产者调用消费者？

【mq】从零开始实现 mq-03- 引入 broker 中间人

【mq】从零开始实现 mq-04- 启动检测与实现优化

上一节咱们引入了中间人 broker，让音讯的生产者和消费者解耦。

这一节咱们对初始化代码进行优化，便于前期拓展保护。

生产者启动优化

启动实现

整体实现调整如下：

@Override
public synchronized void run() {this.paramCheck();
    // 启动服务端
    log.info("MQ 生产者开始启动客户端 GROUP: {}, PORT: {}, brokerAddress: {}",
            groupName, port, brokerAddress);
    try {
        //channel future
        this.channelFutureList = ChannelFutureUtils.initChannelFutureList(brokerAddress,
                initChannelHandler(), check);

        // register to broker
        this.registerToBroker();

        // 标识为可用
        enableFlag = true;
        log.info("MQ 生产者启动实现");
    } catch (Exception e) {log.error("MQ 生产者启动遇到异样", e);
        throw new MqException(ProducerRespCode.RPC_INIT_FAILED);
    }
}

看起来是不是比起原来清新很多呢？

然而 复杂性只会转移，不会隐没。

答案就是封装到 initChannelFutureList 中去了。

initChannelFutureList

因为这里是生产者、消费者都会用到。

所以咱们先放在对立的工具类中，实现自身和以前大同小异。

/**
 * 初始化列表
 * @param brokerAddress 地址
 * @param channelHandler 解决类
 * @param check 是否检测可用性
 * @return 后果
 * @since 0.0.4
 */
public static List<RpcChannelFuture> initChannelFutureList(final String brokerAddress,
                                                           final ChannelHandler channelHandler,
                                                           final boolean check) {List<RpcAddress> addressList = InnerAddressUtils.initAddressList(brokerAddress);
    List<RpcChannelFuture> list = new ArrayList<>();
    for(RpcAddress rpcAddress : addressList) {
        try {final String address = rpcAddress.getAddress();
            final int port = rpcAddress.getPort();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            ChannelFuture channelFuture = bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<Channel>(){
                        @Override
                        protected void initChannel(Channel ch) throws Exception {ch.pipeline()
                                    .addLast(new LoggingHandler(LogLevel.INFO))
                                    .addLast(channelHandler);
                        }
                    })
                    .connect(address, port)
                    .syncUninterruptibly();
            log.info("启动客户端实现，监听 address: {}, port：{}", address, port);
            RpcChannelFuture rpcChannelFuture = new RpcChannelFuture();
            rpcChannelFuture.setChannelFuture(channelFuture);
            rpcChannelFuture.setAddress(address);
            rpcChannelFuture.setPort(port);
            rpcChannelFuture.setWeight(rpcAddress.getWeight());
            list.add(rpcChannelFuture);
        } catch (Exception exception) {log.error("注册到 broker 服务端异样", exception);
            if(check) {throw new MqException(MqCommonRespCode.REGISTER_TO_BROKER_FAILED);
            }
        }
    }

    if(check
        && CollectionUtil.isEmpty(list)) {log.error("check=true 且可用列表为空，启动失败。");
        throw new MqException(MqCommonRespCode.REGISTER_TO_BROKER_FAILED);
    }
    return list;
}

这里的 check 为了防止 2 种状况：

（1）某一个 broker 不可用

（2）没有可用的 broker 信息。

消费者启动优化

消费者连贯 broker 和生产者是相似的。

这里只是放一下实现，不做更多的赘述。

@Override
public void run() {
    // 启动服务端
    log.info("MQ 消费者开始启动服务端 groupName: {}, brokerAddress: {}",
            groupName, brokerAddress);
    //1. 参数校验
    this.paramCheck();
    try {
        //channel future
        this.channelFutureList = ChannelFutureUtils.initChannelFutureList(brokerAddress,
                initChannelHandler(),
                check);

        // register to broker
        this.registerToBroker();

        // 标识为可用
        enableFlag = true;
        log.info("MQ 消费者启动实现");
    } catch (Exception e) {log.error("MQ 消费者启动异样", e);
        throw new MqException(ConsumerRespCode.RPC_INIT_FAILED);
    }
}

小结

这一大节的内容特地简略，对初始化局部做了优化，便于前期保护拓展。

心愿本文对你有所帮忙，如果喜爱，欢送点赞珍藏转发一波。

我是老马，期待与你的下次重逢。

开源地址

The message queue in java.(java 繁难版本 mq 实现) https://github.com/houbb/mq

拓展浏览

rpc- 从零开始实现 rpc https://github.com/houbb/rpc