前景回顾
【mq】从零开始实现 mq-01-生产者、消费者启动
【mq】从零开始实现 mq-02-如何实现生产者调用消费者？
【mq】从零开始实现 mq-03-引入 broker 中间人
【mq】从零开始实现 mq-04-启动检测与实现优化
上一节咱们引入了中间人 broker，让音讯的生产者和消费者解耦。
这一节咱们对初始化代码进行优化，便于前期拓展保护。
生产者启动优化
启动实现
整体实现调整如下：
@Overridepublic 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 和生产者是相似的。
这里只是放一下实现，不做更多的赘述。
@Overridepublic 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