前景回顾
【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
