标签：log 04 启动 broker 从零开始 mq brokerAddress new
前景回顾
【mq】从零开始实现 mq-02-如何实现生产者调用消费者？
【mq】从零开始实现 mq-03-引入 broker 中间人
上一节我们引入了中间人 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
标签：log,04,启动,broker,从零开始,mq,brokerAddress,new 来源： https://www.cnblogs.com/houbbBlogs/p/16218299.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。