关于mq:mq从零开始实现-mq03引入-broker-中间人

41次阅读

共计 22618 个字符，预计需要花费 57 分钟才能阅读完成。

前景回顾

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

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

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

上一节咱们学习了如何实现生产者给消费者发送音讯，然而是通过直连的形式。

那么如何能力达到解耦的成果呢？

答案就是引入 broker，音讯的中间人。

MqBroker 实现

外围启动类

相似咱们后面 consumer 的启动实现：

package com.github.houbb.mq.broker.core;

/**
 * @author binbin.hou
 * @since 1.0.0
 */
public class MqBroker extends Thread implements IMqBroker {

    // 省略
    private ChannelHandler initChannelHandler() {MqBrokerHandler handler = new MqBrokerHandler();
        handler.setInvokeService(invokeService);
        handler.setRegisterConsumerService(registerConsumerService);
        handler.setRegisterProducerService(registerProducerService);
        handler.setMqBrokerPersist(mqBrokerPersist);
        handler.setBrokerPushService(brokerPushService);
        handler.setRespTimeoutMills(respTimeoutMills);

        return handler;
    }

    @Override
    public void run() {
        // 启动服务端
        log.info("MQ 中间人开始启动服务端 port: {}", port);

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {final ByteBuf delimiterBuf = DelimiterUtil.getByteBuf(DelimiterUtil.DELIMITER);
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(workerGroup, bossGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {ch.pipeline()
                                    .addLast(new DelimiterBasedFrameDecoder(DelimiterUtil.LENGTH, delimiterBuf))
                                    .addLast(initChannelHandler());
                        }
                    })
                    // 这个参数影响的是还没有被 accept 取出的连贯
                    .option(ChannelOption.SO_BACKLOG, 128)
                    // 这个参数只是过一段时间内客户端没有响应，服务端会发送一个 ack 包，以判断客户端是否还活着。.childOption(ChannelOption.SO_KEEPALIVE, true);

            // 绑定端口，开始接管进来的链接
            ChannelFuture channelFuture = serverBootstrap.bind(port).syncUninterruptibly();
            log.info("MQ 中间人启动实现，监听【" + port + "】端口");

            channelFuture.channel().closeFuture().syncUninterruptibly();
            log.info("MQ 中间人敞开实现");
        } catch (Exception e) {log.error("MQ 中间人启动异样", e);
            throw new MqException(BrokerRespCode.RPC_INIT_FAILED);
        } finally {workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();}
    }

}

initChannelHandler 中有不少新面孔，咱们前面会具体介绍。

MqBrokerHandler 解决逻辑 

package com.github.houbb.mq.broker.handler;

import java.util.List;

/**
 * @author binbin.hou
 * @since 1.0.0
 */
public class MqBrokerHandler extends SimpleChannelInboundHandler {private static final Log log = LogFactory.getLog(MqBrokerHandler.class);

    /**
     * 调用治理类
     * @since 1.0.0
     */
    private IInvokeService invokeService;

    /**
     * 消费者治理
     * @since 0.0.3
     */
    private IBrokerConsumerService registerConsumerService;

    /**
     * 生产者治理
     * @since 0.0.3
     */
    private IBrokerProducerService registerProducerService;

    /**
     * 长久化类
     * @since 0.0.3
     */
    private IMqBrokerPersist mqBrokerPersist;

    /**
     * 推送服务
     * @since 0.0.3
     */
    private IBrokerPushService brokerPushService;

    /**
     * 获取响应超时工夫
     * @since 0.0.3
     */
    private long respTimeoutMills;

    //set 办法


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = (ByteBuf) msg;
        byte[] bytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(bytes);

        RpcMessageDto rpcMessageDto = null;
        try {rpcMessageDto = JSON.parseObject(bytes, RpcMessageDto.class);
        } catch (Exception exception) {log.error("RpcMessageDto json 格局转换异样 {}", new String(bytes));
            return;
        }

        if (rpcMessageDto.isRequest()) {MqCommonResp commonResp = this.dispatch(rpcMessageDto, ctx);

            if(commonResp == null) {log.debug("以后音讯为 null，疏忽解决。");
                return;
            }

            // 写回响应，和以前相似。writeResponse(rpcMessageDto, commonResp, ctx);
        } else {final String traceId = rpcMessageDto.getTraceId();

            // 抛弃掉 traceId 为空的信息
            if(StringUtil.isBlank(traceId)) {log.debug("[Server Response] response traceId 为空，间接抛弃", JSON.toJSON(rpcMessageDto));
                return;
            }

            // 增加音讯
            invokeService.addResponse(traceId, rpcMessageDto);
        }
    }

    /**
     * 异步解决音讯
     * @param mqMessage 音讯
     * @since 0.0.3
     */
    private void asyncHandleMessage(MqMessage mqMessage) {List<Channel> channelList = registerConsumerService.getSubscribeList(mqMessage);
        if(CollectionUtil.isEmpty(channelList)) {log.info("监听列表为空，疏忽解决");
            return;
        }

        BrokerPushContext brokerPushContext = new BrokerPushContext();
        brokerPushContext.setChannelList(channelList);
        brokerPushContext.setMqMessage(mqMessage);
        brokerPushContext.setMqBrokerPersist(mqBrokerPersist);
        brokerPushContext.setInvokeService(invokeService);
        brokerPushContext.setRespTimeoutMills(respTimeoutMills);

        brokerPushService.asyncPush(brokerPushContext);
    }
}

音讯散发

broker 接管到音讯当前，dispatch 实现如下：

/**
 * 音讯的散发
 *
 * @param rpcMessageDto 入参
 * @param ctx 上下文
 * @return 后果
 */
private MqCommonResp dispatch(RpcMessageDto rpcMessageDto, ChannelHandlerContext ctx) {
    try {final String methodType = rpcMessageDto.getMethodType();
        final String json = rpcMessageDto.getJson();
        String channelId = ChannelUtil.getChannelId(ctx);
        final Channel channel = ctx.channel();
        log.debug("channelId: {} 接管到 method: {} 内容：{}", channelId,
                methodType, json);

        // 生产者注册
        if(MethodType.P_REGISTER.equals(methodType)) {BrokerRegisterReq registerReq = JSON.parseObject(json, BrokerRegisterReq.class);
            return registerProducerService.register(registerReq.getServiceEntry(), channel);
        }
        // 生产者登记
        if(MethodType.P_UN_REGISTER.equals(methodType)) {BrokerRegisterReq registerReq = JSON.parseObject(json, BrokerRegisterReq.class);
            return registerProducerService.unRegister(registerReq.getServiceEntry(), channel);
        }
        // 生产者音讯发送
        if(MethodType.P_SEND_MSG.equals(methodType)) {MqMessage mqMessage = JSON.parseObject(json, MqMessage.class);
            MqMessagePersistPut persistPut = new MqMessagePersistPut();
            persistPut.setMqMessage(mqMessage);
            persistPut.setMessageStatus(MessageStatusConst.WAIT_CONSUMER);
            MqCommonResp commonResp = mqBrokerPersist.put(persistPut);
            this.asyncHandleMessage(mqMessage);
            return commonResp;
        }
        // 生产者音讯发送 -ONE WAY
        if(MethodType.P_SEND_MSG_ONE_WAY.equals(methodType)) {MqMessage mqMessage = JSON.parseObject(json, MqMessage.class);
            MqMessagePersistPut persistPut = new MqMessagePersistPut();
            persistPut.setMqMessage(mqMessage);
            persistPut.setMessageStatus(MessageStatusConst.WAIT_CONSUMER);
            mqBrokerPersist.put(persistPut);
            this.asyncHandleMessage(mqMessage);
            return null;
        }

        // 消费者注册
        if(MethodType.C_REGISTER.equals(methodType)) {BrokerRegisterReq registerReq = JSON.parseObject(json, BrokerRegisterReq.class);
            return registerConsumerService.register(registerReq.getServiceEntry(), channel);
        }
        // 消费者登记
        if(MethodType.C_UN_REGISTER.equals(methodType)) {BrokerRegisterReq registerReq = JSON.parseObject(json, BrokerRegisterReq.class);
            return registerConsumerService.unRegister(registerReq.getServiceEntry(), channel);
        }
        // 消费者监听注册
        if(MethodType.C_SUBSCRIBE.equals(methodType)) {ConsumerSubscribeReq req = JSON.parseObject(json, ConsumerSubscribeReq.class);
            return registerConsumerService.subscribe(req, channel);
        }
        // 消费者监听登记
        if(MethodType.C_UN_SUBSCRIBE.equals(methodType)) {ConsumerUnSubscribeReq req = JSON.parseObject(json, ConsumerUnSubscribeReq.class);
            return registerConsumerService.unSubscribe(req, channel);
        }

        // 消费者被动 pull
        if(MethodType.C_MESSAGE_PULL.equals(methodType)) {MqConsumerPullReq req = JSON.parseObject(json, MqConsumerPullReq.class);
            return mqBrokerPersist.pull(req, channel);
        }
        throw new UnsupportedOperationException("暂不反对的办法类型");
    } catch (Exception exception) {log.error("执行异样", exception);
        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.FAIL.getCode());
        resp.setRespMessage(MqCommonRespCode.FAIL.getMsg());
        return resp;
    }
}

音讯推送

this.asyncHandleMessage(mqMessage); 是 broker 接管到音讯之后的解决逻辑。

/**
 * 异步解决音讯
 * @param mqMessage 音讯
 * @since 0.0.3
 */
private void asyncHandleMessage(MqMessage mqMessage) {List<Channel> channelList = registerConsumerService.getSubscribeList(mqMessage);
    if(CollectionUtil.isEmpty(channelList)) {log.info("监听列表为空，疏忽解决");
        return;
    }

    BrokerPushContext brokerPushContext = new BrokerPushContext();
    brokerPushContext.setChannelList(channelList);
    brokerPushContext.setMqMessage(mqMessage);
    brokerPushContext.setMqBrokerPersist(mqBrokerPersist);
    brokerPushContext.setInvokeService(invokeService);
    brokerPushContext.setRespTimeoutMills(respTimeoutMills);
    brokerPushService.asyncPush(brokerPushContext);
}

推送的外围实现如下：

package com.github.houbb.mq.broker.support.push;

/**
 * @author binbin.hou
 * @since 0.0.3
 */
public class BrokerPushService implements IBrokerPushService {private static final Log log = LogFactory.getLog(BrokerPushService.class);

    private static final ExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadExecutor();

    @Override
    public void asyncPush(final BrokerPushContext context) {EXECUTOR_SERVICE.submit(new Runnable() {
            @Override
            public void run() {log.info("开始异步解决 {}", JSON.toJSON(context));
                final List<Channel> channelList = context.getChannelList();
                final IMqBrokerPersist mqBrokerPersist = context.getMqBrokerPersist();
                final MqMessage mqMessage = context.getMqMessage();
                final String messageId = mqMessage.getTraceId();
                final IInvokeService invokeService = context.getInvokeService();
                final long responseTime = context.getRespTimeoutMills();

                for(Channel channel : channelList) {
                    try {String channelId = ChannelUtil.getChannelId(channel);

                        log.info("开始解决 channelId: {}", channelId);
                        //1. 调用
                        mqMessage.setMethodType(MethodType.B_MESSAGE_PUSH);
                        MqConsumerResultResp resultResp = callServer(channel, mqMessage,
                                MqConsumerResultResp.class, invokeService, responseTime);

                        //2. 更新状态
                        mqBrokerPersist.updateStatus(messageId, resultResp.getConsumerStatus());

                        //3. 前期增加重试策略

                        log.info("实现解决 channelId: {}", channelId);
                    } catch (Exception exception) {log.error("解决异样");
                        mqBrokerPersist.updateStatus(messageId, ConsumerStatus.FAILED.getCode());
                    }
                }

                log.info("实现异步解决");
            }
        });
    }
}

此处在音讯推送之后，须要更新音讯的 ACK 状态。

音讯生产者解决类

IBrokerProducerService 接口定义如下：

package com.github.houbb.mq.broker.api;

/**
 * <p> 生产者注册服务类 </p>
 *
 * @author houbinbin
 * @since 0.0.3
 */
public interface IBrokerProducerService {

    /**
     * 注册以后服务信息
     *（1）将该服务通过 {@link ServiceEntry#getGroupName()} 进行分组
     * 订阅了这个 serviceId 的所有客户端
     * @param serviceEntry 注册以后服务信息
     * @param channel channel
     * @since 0.0.8
     */
    MqCommonResp register(final ServiceEntry serviceEntry, Channel channel);

    /**
     * 登记以后服务信息
     * @param serviceEntry 注册以后服务信息
     * @param channel 通道
     * @since 0.0.8
     */
    MqCommonResp unRegister(final ServiceEntry serviceEntry, Channel channel);

    /**
     * 获取服务地址信息
     * @param channel channel
     * @return 后果
     * @since 0.0.3
     */
    ServiceEntry getServiceEntry(final Channel channel);

}

实现如下：

本地基于 map 存储申请过去的根本信息。

package com.github.houbb.mq.broker.support.api;

/**
 * <p> 生产者注册服务类 </p>
 *
 * @author houbinbin
 * @since 0.0.3
 */
public class LocalBrokerProducerService implements IBrokerProducerService {private static final Log log = LogFactory.getLog(LocalBrokerProducerService.class);

    private final Map<String, BrokerServiceEntryChannel> registerMap = new ConcurrentHashMap<>();

    @Override
    public MqCommonResp register(ServiceEntry serviceEntry, Channel channel) {final String channelId = ChannelUtil.getChannelId(channel);
        BrokerServiceEntryChannel entryChannel = InnerChannelUtils.buildEntryChannel(serviceEntry, channel);
        registerMap.put(channelId, entryChannel);


        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }

    @Override
    public MqCommonResp unRegister(ServiceEntry serviceEntry, Channel channel) {final String channelId = ChannelUtil.getChannelId(channel);
        registerMap.remove(channelId);

        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }

    @Override
    public ServiceEntry getServiceEntry(Channel channel) {final String channelId = ChannelUtil.getChannelId(channel);
        return registerMap.get(channelId);
    }

}

音讯消费者解决类

接口定义如下：

package com.github.houbb.mq.broker.api;

/**
 * <p> 消费者注册服务类 </p>
 *
 * @author houbinbin
 * @since 0.0.3
 */
public interface IBrokerConsumerService {

    /**
     * 注册以后服务信息
     *（1）将该服务通过 {@link ServiceEntry#getGroupName()} 进行分组
     * 订阅了这个 serviceId 的所有客户端
     * @param serviceEntry 注册以后服务信息
     * @param channel channel
     * @since 0.0.3
     */
    MqCommonResp register(final ServiceEntry serviceEntry, Channel channel);

    /**
     * 登记以后服务信息
     * @param serviceEntry 注册以后服务信息
     * @param channel channel
     * @since 0.0.3
     */
    MqCommonResp unRegister(final ServiceEntry serviceEntry, Channel channel);

    /**
     * 监听服务信息
     *（1）监听之后，如果有任何相干的机器信息发生变化，则进行推送。*（2）内置的信息，须要传送 ip 信息到注册核心。*
     * @param serviceEntry 客户端明细信息
     * @param clientChannel 客户端 channel 信息
     * @since 0.0.3
     */
    MqCommonResp subscribe(final ConsumerSubscribeReq serviceEntry,
                   final Channel clientChannel);

    /**
     * 勾销监听服务信息
     *（1）监听之后，如果有任何相干的机器信息发生变化，则进行推送。*（2）内置的信息，须要传送 ip 信息到注册核心。*
     * @param serviceEntry 客户端明细信息
     * @param clientChannel 客户端 channel 信息
     * @since 0.0.3
     */
    MqCommonResp unSubscribe(final ConsumerUnSubscribeReq serviceEntry,
                   final Channel clientChannel);

    /**
     * 获取所有匹配的消费者
     * 1. 同一个 groupName 只返回一个，留神负载平衡
     * 2. 返回匹配以后音讯的消费者通道
     *
     * @param mqMessage 音讯体
     * @return 后果
     */
    List<Channel> getSubscribeList(MqMessage mqMessage);

}

默认实现：

package com.github.houbb.mq.broker.support.api;

/**
 * @author binbin.hou
 * @since 1.0.0
 */
public class LocalBrokerConsumerService implements IBrokerConsumerService {private final Map<String, BrokerServiceEntryChannel> registerMap = new ConcurrentHashMap<>();

    /**
     * 订阅汇合
     * key: topicName
     * value: 对应的订阅列表
     */
    private final Map<String, Set<ConsumerSubscribeBo>> subscribeMap = new ConcurrentHashMap<>();

    @Override
    public MqCommonResp register(ServiceEntry serviceEntry, Channel channel) {final String channelId = ChannelUtil.getChannelId(channel);
        BrokerServiceEntryChannel entryChannel = InnerChannelUtils.buildEntryChannel(serviceEntry, channel);
        registerMap.put(channelId, entryChannel);

        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }

    @Override
    public MqCommonResp unRegister(ServiceEntry serviceEntry, Channel channel) {final String channelId = ChannelUtil.getChannelId(channel);
        registerMap.remove(channelId);

        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }

    @Override
    public MqCommonResp subscribe(ConsumerSubscribeReq serviceEntry, Channel clientChannel) {final String channelId = ChannelUtil.getChannelId(clientChannel);
        final String topicName = serviceEntry.getTopicName();

        Set<ConsumerSubscribeBo> set = subscribeMap.get(topicName);
        if(set == null) {set = new HashSet<>();
        }
        ConsumerSubscribeBo subscribeBo = new ConsumerSubscribeBo();
        subscribeBo.setChannelId(channelId);
        subscribeBo.setGroupName(serviceEntry.getGroupName());
        subscribeBo.setTopicName(topicName);
        subscribeBo.setTagRegex(serviceEntry.getTagRegex());
        set.add(subscribeBo);

        subscribeMap.put(topicName, set);

        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }

    @Override
    public MqCommonResp unSubscribe(ConsumerUnSubscribeReq serviceEntry, Channel clientChannel) {final String channelId = ChannelUtil.getChannelId(clientChannel);
        final String topicName = serviceEntry.getTopicName();

        ConsumerSubscribeBo subscribeBo = new ConsumerSubscribeBo();
        subscribeBo.setChannelId(channelId);
        subscribeBo.setGroupName(serviceEntry.getGroupName());
        subscribeBo.setTopicName(topicName);
        subscribeBo.setTagRegex(serviceEntry.getTagRegex());

        // 汇合
        Set<ConsumerSubscribeBo> set = subscribeMap.get(topicName);
        if(CollectionUtil.isNotEmpty(set)) {set.remove(subscribeBo);
        }

        MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }

    @Override
    public List<Channel> getSubscribeList(MqMessage mqMessage) {final String topicName = mqMessage.getTopic();
        Set<ConsumerSubscribeBo> set = subscribeMap.get(topicName);
        if(CollectionUtil.isEmpty(set)) {return Collections.emptyList();
        }

        //2. 获取匹配的 tag 列表
        final List<String> tagNameList = mqMessage.getTags();

        Map<String, List<ConsumerSubscribeBo>> groupMap = new HashMap<>();
        for(ConsumerSubscribeBo bo : set) {String tagRegex = bo.getTagRegex();

            if(hasMatch(tagNameList, tagRegex)) {
                //TODO: 这种设置模式，对立增加解决
                String groupName = bo.getGroupName();
                List<ConsumerSubscribeBo> list = groupMap.get(groupName);
                if(list == null) {list = new ArrayList<>();
                }
                list.add(bo);

                groupMap.put(groupName, list);
            }
        }

        //3. 依照 groupName 分组之后，每一组只随机返回一个。最好应该调整为以 shardingkey 抉择
        final String shardingKey = mqMessage.getShardingKey();
        List<Channel> channelList = new ArrayList<>();

        for(Map.Entry<String, List<ConsumerSubscribeBo>> entry : groupMap.entrySet()) {List<ConsumerSubscribeBo> list = entry.getValue();

            ConsumerSubscribeBo bo = RandomUtils.random(list, shardingKey);
            BrokerServiceEntryChannel entryChannel = registerMap.get(bo.getChannelId());
            channelList.add(entryChannel.getChannel());
        }

        return channelList;
    }

    private boolean hasMatch(List<String> tagNameList,
                             String tagRegex) {if(CollectionUtil.isEmpty(tagNameList)) {return false;}

        Pattern pattern = Pattern.compile(tagRegex);

        for(String tagName : tagNameList) {if(RegexUtils.match(pattern, tagName)) {return true;}
        }

        return false;
    }

}

getSubscribeList 的逻辑可能略微简单点，其实就是音讯过去，找到匹配的订阅消费者而已。

因为同一个 groupName 的消费者音讯只生产一次，所以须要一次分组。

音讯长久化

接口如下：

package com.github.houbb.mq.broker.support.persist;

/**
 * @author binbin.hou
 * @since 0.0.3
 */
public interface IMqBrokerPersist {

    /**
     * 保留音讯
     * @param mqMessage 音讯
     * @since 0.0.3
     */
    MqCommonResp put(final MqMessagePersistPut mqMessage);

    /**
     * 更新状态
     * @param messageId 音讯惟一标识
     * @param status 状态
     * @return 后果
     * @since 0.0.3
     */
    MqCommonResp updateStatus(final String messageId,
                              final String status);

    /**
     * 拉取音讯
     * @param pull 拉取音讯
     * @return 后果
     */
    MqConsumerPullResp pull(final MqConsumerPullReq pull, final Channel channel);

}

本地默认实现：

package com.github.houbb.mq.broker.support.persist;

/**
 * 本地长久化策略
 * @author binbin.hou
 * @since 1.0.0
 */
public class LocalMqBrokerPersist implements IMqBrokerPersist {private static final Log log = LogFactory.getLog(LocalMqBrokerPersist.class);

    /**
     * 队列
     * ps: 这里只是简化实现，临时不思考并发等问题。*/
    private final Map<String, List<MqMessagePersistPut>> map = new ConcurrentHashMap<>();

    //1. 接管
    //2. 长久化
    //3. 告诉生产
    @Override
    public synchronized MqCommonResp put(MqMessagePersistPut put) {log.info("put elem: {}", JSON.toJSON(put));

        MqMessage mqMessage = put.getMqMessage();
        final String topic = mqMessage.getTopic();

        List<MqMessagePersistPut> list = map.get(topic);
        if(list == null) {list = new ArrayList<>();
        }
        list.add(put);
        map.put(topic, list);

        MqCommonResp commonResp = new MqCommonResp();
        commonResp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        commonResp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return commonResp;
    }

    @Override
    public MqCommonResp updateStatus(String messageId, String status) {
        // 这里性能比拟差，所以不能够用于生产。仅作为测试验证
        for(List<MqMessagePersistPut> list : map.values()) {for(MqMessagePersistPut put : list) {MqMessage mqMessage = put.getMqMessage();
                if(mqMessage.getTraceId().equals(messageId)) {put.setMessageStatus(status);

                    break;
                }
            }
        }

        MqCommonResp commonResp = new MqCommonResp();
        commonResp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        commonResp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return commonResp;
    }

    @Override
    public MqConsumerPullResp pull(MqConsumerPullReq pull, Channel channel) {
        //TODO... 待实现
        return null;
    }

}

ps: 后续将会基于 springboot+mysql 进行长久化策略实现。

消费者启动调整

咱们将生产者、消费者的启动都进行调整，连贯到 broker 中。

二者是相似的，此处以消费者为例。

外围启动类 

package com.github.houbb.mq.consumer.core;

/**
 * 推送生产策略
 *
 * @author binbin.hou
 * @since 1.0.0
 */
public class MqConsumerPush extends Thread implements IMqConsumer  {

    // 属性 & 设置

    @Override
    public void run() {
        // 启动服务端
        log.info("MQ 消费者开始启动服务端 groupName: {}, brokerAddress: {}",
                groupName, brokerAddress);

        //1. 参数校验
        this.paramCheck();

        try {
            // channel handler
            ChannelHandler channelHandler = this.initChannelHandler();

            //channel future
            this.channelFutureList = ChannelFutureUtils.initChannelFutureList(brokerAddress, channelHandler);

            // register to broker
            this.registerToBroker();

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

    // 订阅 & 勾销订阅

    @Override
    public void registerListener(IMqConsumerListener listener) {this.mqListenerService.register(listener);
    }

}

初始化 handler

private ChannelHandler initChannelHandler() {final ByteBuf delimiterBuf = DelimiterUtil.getByteBuf(DelimiterUtil.DELIMITER);

    final MqConsumerHandler mqConsumerHandler = new MqConsumerHandler(invokeService, mqListenerService);
    // handler 实际上会被屡次调用，如果不是 @Shareable，应该每次都从新创立。ChannelHandler handler = new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {ch.pipeline()
                    .addLast(new DelimiterBasedFrameDecoder(DelimiterUtil.LENGTH, delimiterBuf))
                    .addLast(mqConsumerHandler);
        }
    };

    return handler;
}

注册到服务端 

/**
 * 注册到所有的服务端
 * @since 0.0.3
 */
private void registerToBroker() {for(RpcChannelFuture channelFuture : this.channelFutureList) {ServiceEntry serviceEntry = new ServiceEntry();
        serviceEntry.setGroupName(groupName);
        serviceEntry.setAddress(channelFuture.getAddress());
        serviceEntry.setPort(channelFuture.getPort());
        serviceEntry.setWeight(channelFuture.getWeight());

        BrokerRegisterReq brokerRegisterReq = new BrokerRegisterReq();
        brokerRegisterReq.setServiceEntry(serviceEntry);
        brokerRegisterReq.setMethodType(MethodType.C_REGISTER);
        brokerRegisterReq.setTraceId(IdHelper.uuid32());

        log.info("[Register] 开始注册到 broker：{}", JSON.toJSON(brokerRegisterReq));
        final Channel channel = channelFuture.getChannelFuture().channel();
        MqCommonResp resp = callServer(channel, brokerRegisterReq, MqCommonResp.class);
        log.info("[Register] 实现注册到 broker：{}", JSON.toJSON(resp));
    }
}

订阅与勾销订阅

消费者对于关怀的音讯，实现也比较简单：

public void subscribe(String topicName, String tagRegex) {ConsumerSubscribeReq req = new ConsumerSubscribeReq();
    String messageId = IdHelper.uuid32();
    req.setTraceId(messageId);
    req.setMethodType(MethodType.C_SUBSCRIBE);
    req.setTopicName(topicName);
    req.setTagRegex(tagRegex);
    req.setGroupName(groupName);

    Channel channel = getChannel();

    MqCommonResp resp = callServer(channel, req, MqCommonResp.class);
    if(!MqCommonRespCode.SUCCESS.getCode().equals(resp.getRespCode())) {throw new MqException(ConsumerRespCode.SUBSCRIBE_FAILED);
    }
}

勾销订阅：

public void unSubscribe(String topicName, String tagRegex) {ConsumerUnSubscribeReq req = new ConsumerUnSubscribeReq();
    String messageId = IdHelper.uuid32();
    req.setTraceId(messageId);
    req.setMethodType(MethodType.C_UN_SUBSCRIBE);
    req.setTopicName(topicName);
    req.setTagRegex(tagRegex);
    req.setGroupName(groupName);

    Channel channel = getChannel();

    MqCommonResp resp = callServer(channel, req, MqCommonResp.class);
    if(!MqCommonRespCode.SUCCESS.getCode().equals(resp.getRespCode())) {throw new MqException(ConsumerRespCode.UN_SUBSCRIBE_FAILED);
    }
}

测试

broker 启动 

MqBroker broker = new MqBroker();
broker.start();

启动日志：

[DEBUG] [2022-04-21 20:36:27.158] [main] [c.g.h.l.i.c.LogFactory.setImplementation] - Logging initialized using 'class com.github.houbb.log.integration.adaptors.stdout.StdOutExImpl' adapter.
[INFO] [2022-04-21 20:36:27.186] [Thread-0] [c.g.h.m.b.c.MqBroker.run] - MQ 中间人开始启动服务端 port: 9999
[INFO] [2022-04-21 20:36:29.060] [Thread-0] [c.g.h.m.b.c.MqBroker.run] - MQ 中间人启动实现，监听【9999】端口

consumer 启动 

final MqConsumerPush mqConsumerPush = new MqConsumerPush();
mqConsumerPush.start();

mqConsumerPush.subscribe("TOPIC", "TAGA");
mqConsumerPush.registerListener(new IMqConsumerListener() {
    @Override
    public ConsumerStatus consumer(MqMessage mqMessage, IMqConsumerListenerContext context) {System.out.println("---------- 自定义" + JSON.toJSONString(mqMessage));
        return ConsumerStatus.SUCCESS;
    }
});

启动日志：

...
[INFO] [2022-04-21 20:37:40.985] [Thread-0] [c.g.h.m.c.c.MqConsumerPush.registerToBroker] - [Register] 实现注册到 broker：{"respMessage":"胜利","respCode":"0000"}

启动时会注册到 broker。

producer 启动 

MqProducer mqProducer = new MqProducer();
mqProducer.start();
String message = "HELLO MQ!";
MqMessage mqMessage = new MqMessage();
mqMessage.setTopic("TOPIC");
mqMessage.setTags(Arrays.asList("TAGA", "TAGB"));
mqMessage.setPayload(message);

SendResult sendResult = mqProducer.send(mqMessage);

System.out.println(JSON.toJSON(sendResult));

日志：

...
[INFO] [2022-04-21 20:39:17.885] [Thread-0] [c.g.h.m.p.c.MqProducer.registerToBroker] - [Register] 实现注册到 broker：{"respMessage":"胜利","respCode":"0000"}
...

此时消费者生产到咱们发送的音讯。

---------- 自定义 {"methodType":"B_MESSAGE_PUSH","payload":"HELLO MQ!","tags":["TAGA","TAGB"],"topic":"TOPIC","traceId":"2237bbfe55b842328134e6a100e36364"}

小结

到这里，咱们就实现了基于中间人的生产者与消费者通信。

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

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

开源地址

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

拓展浏览

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

正文完
mq
发表至： mq
2022-04-30
 0
关于mq:MQ-面试要点
关于mq:网关交易同步优化
关于mq:mq从零开始实现-mq05实现优雅停机
关于mq:系统学习消息队列RabbitMQ的消息应答和持久化