关于mq:MQjava-从零开始实现消息队列-mq02如何实现生产者调用消费者

上一节咱们学习了如何实现基于 netty 客服端和服务端的启动。

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

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

那么客户端如何调用服务端呢？

咱们本节就来一起实现一下。

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(new MqConsumerHandler(invokeService));
            }
        })
        // 这个参数影响的是还没有被 accept 取出的连贯
        .option(ChannelOption.SO_BACKLOG, 128)
        // 这个参数只是过一段时间内客户端没有响应，服务端会发送一个 ack 包，以判断客户端是否还活着。.childOption(ChannelOption.SO_KEEPALIVE, true);

这里咱们通过指定分隔符解决 netty 粘包问题。

解决 netty 粘包问题

MqConsumerHandler 的实现如下，增加对应的业务解决逻辑。

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

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

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

    public MqConsumerHandler(IInvokeService invokeService) {this.invokeService = invokeService;}

    @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);
        }
    }
}

为了统一标准，咱们的 rpc 音讯体 RpcMessageDto 定义如下：

package com.github.houbb.mq.common.rpc;

/**
 * @author binbin.hou
 * @since 1.0.0
 */
public class RpcMessageDto implements Serializable {

    /**
     * 申请工夫
     */
    private long requestTime;

    /**
     * 申请标识
     */
    private String traceId;

    /**
     * 办法类型
     */
    private String methodType;

    /**
     * 是否为申请音讯
     */
    private boolean isRequest;

    private String respCode;

    private String respMsg;

    private String json;

    //getter&setter

}

对于接管到的音讯体 RpcMessageDto，散发逻辑如下：

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

    // 音讯发送
    if(MethodType.P_SEND_MESSAGE.equals(methodType)) {
        // 日志输入
        log.info("收到服务端音讯: {}", json);
        // 如果是 broker，应该进行解决化等操作。MqCommonResp resp = new MqCommonResp();
        resp.setRespCode(MqCommonRespCode.SUCCESS.getCode());
        resp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());
        return resp;
    }
    throw new UnsupportedOperationException("暂不反对的办法类型");
}

这里对于接管到的音讯，只做一个简略的日志输入，后续将增加对应的业务逻辑解决。

收到申请当前，咱们须要返回对应的响应。

基于 channel 的回写实现如下：

/**
 * 后果写回
 *
 * @param req  申请
 * @param resp 响应
 * @param ctx  上下文
 */
private void writeResponse(RpcMessageDto req,
                           Object resp,
                           ChannelHandlerContext ctx) {final String id = ctx.channel().id().asLongText();
    RpcMessageDto rpcMessageDto = new RpcMessageDto();
    // 响应类音讯
    rpcMessageDto.setRequest(false);
    rpcMessageDto.setTraceId(req.getTraceId());
    rpcMessageDto.setMethodType(req.getMethodType());
    rpcMessageDto.setRequestTime(System.currentTimeMillis());
    String json = JSON.toJSONString(resp);
    rpcMessageDto.setJson(json);
    // 回写到 client 端
    ByteBuf byteBuf = DelimiterUtil.getMessageDelimiterBuffer(rpcMessageDto);
    ctx.writeAndFlush(byteBuf);
    log.debug("[Server] channel {} response {}", id, JSON.toJSON(rpcMessageDto));
}

为了方便管理异步返回的申请后果，咱们对立定义了 IInvokeService 类，用于治理申请与响应。

package com.github.houbb.mq.common.support.invoke;

import com.github.houbb.mq.common.rpc.RpcMessageDto;

/**
 * 调用服务接口
 * @author binbin.hou
 * @since 1.0.0
 */
public interface IInvokeService {

    /**
     * 增加申请信息
     * @param seqId 序列号
     * @param timeoutMills 超时工夫
     * @return this
     * @since 1.0.0
     */
    IInvokeService addRequest(final String seqId,
                              final long timeoutMills);

    /**
     * 放入后果
     * @param seqId 惟一标识
     * @param rpcResponse 响应后果
     * @return this
     * @since 1.0.0
     */
    IInvokeService addResponse(final String seqId, final RpcMessageDto rpcResponse);

    /**
     * 获取标记信息对应的后果
     * @param seqId 序列号
     * @return 后果
     * @since 1.0.0
     */
    RpcMessageDto getResponse(final String seqId);

}

实现自身也不难。

package com.github.houbb.mq.common.support.invoke.impl;

/**
 * 调用服务接口
 * @author binbin.hou
 * @since 1.0.0
 */
public class InvokeService implements IInvokeService {private static final Log logger = LogFactory.getLog(InvokeService.class);

    /**
     * 申请序列号 map
     *（1）这里前期如果要增加超时检测，能够增加对应的超时工夫。* 能够把这里调整为 map
     *
     * key: seqId 惟一标识一个申请
     * value: 存入该申请最长的无效工夫。用于定时删除和超时判断。* @since 0.0.2
     */
    private final ConcurrentHashMap<String, Long> requestMap;

    /**
     * 响应后果
     * @since 1.0.0
     */
    private final ConcurrentHashMap<String, RpcMessageDto> responseMap;

    public InvokeService() {requestMap = new ConcurrentHashMap<>();
        responseMap = new ConcurrentHashMap<>();

        final Runnable timeoutThread = new TimeoutCheckThread(requestMap, responseMap);
        Executors.newScheduledThreadPool(1)
                .scheduleAtFixedRate(timeoutThread,60, 60, TimeUnit.SECONDS);
    }

    @Override
    public IInvokeService addRequest(String seqId, long timeoutMills) {logger.debug("[Invoke] start add request for seqId: {}, timeoutMills: {}", seqId,
                timeoutMills);

        final long expireTime = System.currentTimeMillis()+timeoutMills;
        requestMap.putIfAbsent(seqId, expireTime);

        return this;
    }

    @Override
    public IInvokeService addResponse(String seqId, RpcMessageDto rpcResponse) {
        // 1. 判断是否无效
        Long expireTime = this.requestMap.get(seqId);
        // 如果为空，可能是这个后果曾经超时了，被定时 job 移除之后，响应后果才过去。间接疏忽
        if(ObjectUtil.isNull(expireTime)) {return this;}

        //2. 判断是否超时
        if(System.currentTimeMillis() > expireTime) {logger.debug("[Invoke] seqId:{} 信息已超时，间接返回超时后果。", seqId);
            rpcResponse = RpcMessageDto.timeout();}

        // 这里放入之前，能够增加判断。// 如果 seqId 必须解决申请汇合中，才容许放入。或者间接疏忽抛弃。// 告诉所有期待方
        responseMap.putIfAbsent(seqId, rpcResponse);
        logger.debug("[Invoke] 获取后果信息，seqId: {}, rpcResponse: {}", seqId, JSON.toJSON(rpcResponse));
        logger.debug("[Invoke] seqId:{} 信息曾经放入，告诉所有期待方", seqId);

        // 移除对应的 requestMap
        requestMap.remove(seqId);
        logger.debug("[Invoke] seqId:{} remove from request map", seqId);

        // 同步锁
        synchronized (this) {this.notifyAll();
            logger.debug("[Invoke] {} notifyAll()", seqId);
        }


        return this;
    }

    @Override
    public RpcMessageDto getResponse(String seqId) {
        try {RpcMessageDto rpcResponse = this.responseMap.get(seqId);
            if(ObjectUtil.isNotNull(rpcResponse)) {logger.debug("[Invoke] seq {} 对应后果曾经获取: {}", seqId, rpcResponse);
                return rpcResponse;
            }

            // 进入期待
            while (rpcResponse == null) {logger.debug("[Invoke] seq {} 对应后果为空，进入期待", seqId);

                // 同步期待锁
                synchronized (this) {this.wait();
                }

                logger.debug("[Invoke] {} wait has notified!", seqId);

                rpcResponse = this.responseMap.get(seqId);
                logger.debug("[Invoke] seq {} 对应后果曾经获取: {}", seqId, rpcResponse);
            }

            return rpcResponse;
        } catch (InterruptedException e) {logger.error("获取响应异样", e);
            throw new MqException(MqCommonRespCode.RPC_GET_RESP_FAILED);
        }
    }

}

这里 getResponse 获取不到会进入期待，直到 addResponse 唤醒。

然而这也有一个问题，如果一个申请的响应失落了怎么办？

总不能始终期待吧。

超时检测线程就能够帮咱们解决一些超时未返回的后果。

package com.github.houbb.mq.common.support.invoke.impl;

import com.github.houbb.heaven.util.common.ArgUtil;
import com.github.houbb.mq.common.rpc.RpcMessageDto;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 超时检测线程
 * @author binbin.hou
 * @since 0.0.2
 */
public class TimeoutCheckThread implements Runnable {

    /**
     * 申请信息
     * @since 0.0.2
     */
    private final ConcurrentHashMap<String, Long> requestMap;

    /**
     * 申请信息
     * @since 0.0.2
     */
    private final ConcurrentHashMap<String, RpcMessageDto> responseMap;

    /**
     * 新建
     * @param requestMap  申请 Map
     * @param responseMap 后果 map
     * @since 0.0.2
     */
    public TimeoutCheckThread(ConcurrentHashMap<String, Long> requestMap,
                              ConcurrentHashMap<String, RpcMessageDto> responseMap) {ArgUtil.notNull(requestMap, "requestMap");
        this.requestMap = requestMap;
        this.responseMap = responseMap;
    }

    @Override
    public void run() {for(Map.Entry<String, Long> entry : requestMap.entrySet()) {long expireTime = entry.getValue();
            long currentTime = System.currentTimeMillis();

            if(currentTime > expireTime) {final String key = entry.getKey();
                // 后果设置为超时，从申请 map 中移除
                responseMap.putIfAbsent(key, RpcMessageDto.timeout());
                requestMap.remove(key);
            }
        }
    }

}

解决逻辑就是定时检测，如果超时了，就默认设置后果为超时，并且从申请汇合中移除。

public class MqProducer extends Thread implements IMqProducer {private static final Log log = LogFactory.getLog(MqProducer.class);

    /**
     * 分组名称
     */
    private final String groupName;

    /**
     * 端口号
     */
    private final int port;

    /**
     * 中间人地址
     */
    private String brokerAddress  = "";

    /**
     * channel 信息
     * @since 0.0.2
     */
    private ChannelFuture channelFuture;

    /**
     * 客户端解决 handler
     * @since 0.0.2
     */
    private ChannelHandler channelHandler;

    /**
     * 调用治理服务
     * @since 0.0.2
     */
    private final IInvokeService invokeService = new InvokeService();

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

    /**
     * 可用标识
     * @since 0.0.2
     */
    private volatile boolean enableFlag = false;

    /**
     * 粘包解决分隔符
     * @since 1.0.0
     */
    private String delimiter = DelimiterUtil.DELIMITER;

    //set 办法

    

    @Override
    public synchronized void run() {
        // 启动服务端
        log.info("MQ 生产者开始启动客户端 GROUP: {}, PORT: {}, brokerAddress: {}",
                groupName, port, brokerAddress);

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // channel handler
            this.initChannelHandler();

            // 省略，同以前

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

}

其中初始化 handler 的实现如下：

private void initChannelHandler() {final ByteBuf delimiterBuf = DelimiterUtil.getByteBuf(delimiter);

    final MqProducerHandler mqProducerHandler = new MqProducerHandler();
    mqProducerHandler.setInvokeService(invokeService);

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

和消费者解决逻辑相似。

这里最外围的就是增加响应后果：invokeService.addResponse(rpcMessageDto.getTraceId(), rpcMessageDto);

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

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

    /**
     * 调用治理类
     */
    private IInvokeService invokeService;

    public void setInvokeService(IInvokeService invokeService) {this.invokeService = invokeService;}

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

        String text = new String(bytes);
        log.debug("[Client] channelId {} 接管到音讯 {}", ChannelUtil.getChannelId(ctx), text);

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

        if(rpcMessageDto.isRequest()) {
            // 申请类
            final String methodType = rpcMessageDto.getMethodType();
            final String json = rpcMessageDto.getJson();} else {
            // 抛弃掉 traceId 为空的信息
            if(StringUtil.isBlank(rpcMessageDto.getTraceId())) {log.debug("[Client] response traceId 为空，间接抛弃", JSON.toJSON(rpcMessageDto));
                return;
            }

            invokeService.addResponse(rpcMessageDto.getTraceId(), rpcMessageDto);
            log.debug("[Client] response is :{}", JSON.toJSON(rpcMessageDto));
        }
    }
}

关怀申请后果的：

public SendResult send(MqMessage mqMessage) {String messageId = IdHelper.uuid32();
    mqMessage.setTraceId(messageId);
    mqMessage.setMethodType(MethodType.P_SEND_MESSAGE);
    MqCommonResp resp = callServer(mqMessage, MqCommonResp.class);
    if(MqCommonRespCode.SUCCESS.getCode().equals(resp.getRespCode())) {return SendResult.of(messageId, SendStatus.SUCCESS);
    }
    return SendResult.of(messageId, SendStatus.FAILED);
}

不关怀申请后果的发送：

public SendResult sendOneWay(MqMessage mqMessage) {String messageId = IdHelper.uuid32();
    mqMessage.setTraceId(messageId);
    mqMessage.setMethodType(MethodType.P_SEND_MESSAGE);
    this.callServer(mqMessage, null);
    return SendResult.of(messageId, SendStatus.SUCCESS);
}

其中 callServer 实现如下：

/**
 * 调用服务端
 * @param commonReq 通用申请
 * @param respClass 类
 * @param <T> 泛型
 * @param <R> 后果
 * @return 后果
 * @since 1.0.0
 */
public <T extends MqCommonReq, R extends MqCommonResp> R callServer(T commonReq, Class<R> respClass) {final String traceId = commonReq.getTraceId();
    final long requestTime = System.currentTimeMillis();
    RpcMessageDto rpcMessageDto = new RpcMessageDto();
    rpcMessageDto.setTraceId(traceId);
    rpcMessageDto.setRequestTime(requestTime);
    rpcMessageDto.setJson(JSON.toJSONString(commonReq));
    rpcMessageDto.setMethodType(commonReq.getMethodType());
    rpcMessageDto.setRequest(true);
    // 增加调用服务
    invokeService.addRequest(traceId, respTimeoutMills);

    // 遍历 channel
    // 敞开以后线程，以获取对应的信息
    // 应用序列化的形式
    ByteBuf byteBuf = DelimiterUtil.getMessageDelimiterBuffer(rpcMessageDto);
    // 负载平衡获取 channel
    Channel channel = channelFuture.channel();
    channel.writeAndFlush(byteBuf);
    String channelId = ChannelUtil.getChannelId(channel);

    log.debug("[Client] channelId {} 发送音讯 {}", channelId, JSON.toJSON(rpcMessageDto));
    if (respClass == null) {log.debug("[Client] 以后音讯为 one-way 音讯，疏忽响应");
        return null;
    } else {
        //channelHandler 中获取对应的响应
        RpcMessageDto messageDto = invokeService.getResponse(traceId);
        if (MqCommonRespCode.TIMEOUT.getCode().equals(messageDto.getRespCode())) {throw new MqException(MqCommonRespCode.TIMEOUT);
        }
        String respJson = messageDto.getJson();
        return JSON.parseObject(respJson, respClass);
    }
}

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

启动日志如下：

[DEBUG] [2022-04-21 19:55:26.346] [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 19:55:26.369] [Thread-0] [c.g.h.m.c.c.MqConsumerPush.run] - MQ 消费者开始启动服务端 groupName: C_DEFAULT_GROUP_NAME, port: 9527, brokerAddress: 
[INFO] [2022-04-21 19:55:27.845] [Thread-0] [c.g.h.m.c.c.MqConsumerPush.run] - MQ 消费者启动实现，监听【9527】端口

MqProducer mqProducer = new MqProducer();
mqProducer.start();

// 期待启动实现
while (!mqProducer.isEnableFlag()) {System.out.println("期待初始化实现...");
    DateUtil.sleep(100);
}

String message = "HELLO MQ!";
MqMessage mqMessage = new MqMessage();
mqMessage.setTopic("TOPIC");
mqMessage.setTags(Arrays.asList("TAGA", "TAGB"));
mqMessage.setPayload(message.getBytes(StandardCharsets.UTF_8));

SendResult sendResult = mqProducer.send(mqMessage);
System.out.println(JSON.toJSON(sendResult));

生产者日志：

[INFO] [2022-04-21 19:56:39.609] [Thread-0] [c.g.h.m.p.c.MqProducer.run] - MQ 生产者启动客户端实现，监听端口：9527
...
[DEBUG] [2022-04-21 19:56:39.895] [main] [c.g.h.m.c.s.i.i.InvokeService.addRequest] - [Invoke] start add request for seqId: a70ea2c4325641d6a5b198323228dc24, timeoutMills: 5000
...
[DEBUG] [2022-04-21 19:56:40.282] [main] [c.g.h.m.c.s.i.i.InvokeService.getResponse] - [Invoke] seq a70ea2c4325641d6a5b198323228dc24 对应后果曾经获取: com.github.houbb.mq.common.rpc.RpcMessageDto@a8f0b4
...
{"messageId":"a70ea2c4325641d6a5b198323228dc24","status":"SUCCESS"}

消费者日志：

[DEBUG] [2022-04-21 19:56:40.179] [nioEventLoopGroup-2-1] [c.g.h.m.c.h.MqConsumerHandler.dispatch] - channelId: 502b73fffec4485c-00003954-00000001-384d194f6233433e-c8246542 接管到 method: P_SEND_MESSAGE 内容：{"methodType":"P_SEND_MESSAGE","payload":"SEVMTE8gTVEh","tags":["TAGA","TAGB"],"topic":"TOPIC","traceId":"a70ea2c4325641d6a5b198323228dc24"}

[INFO] [2022-04-21 19:56:40.180] [nioEventLoopGroup-2-1] [c.g.h.m.c.h.MqConsumerHandler.dispatch] - 收到服务端音讯: {"methodType":"P_SEND_MESSAGE","payload":"SEVMTE8gTVEh","tags":["TAGA","TAGB"],"topic":"TOPIC","traceId":"a70ea2c4325641d6a5b198323228dc24"}

[DEBUG] [2022-04-21 19:56:40.234] [nioEventLoopGroup-2-1] [c.g.h.m.c.h.MqConsumerHandler.writeResponse] - [Server] channel 502b73fffec4485c-00003954-00000001-384d194f6233433e-c8246542 response {"requestTime":1650542200182,"traceId":"a70ea2c4325641d6a5b198323228dc24","request":false,"methodType":"P_SEND_MESSAGE","json":"{\"respCode\":\"0000\",\"respMessage\":\" 胜利 \"}"}

能够看到消费者胜利的获取到了生产者的音讯。

到这里，咱们就实现了一个音讯生产者调用消费者的实现。

然而你可能会问，这不就是 rpc 吗？

没有解耦。

是的，为了解决耦合问题，咱们将在下一节引入 broker 音讯的中间人。

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

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

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

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

关于mq:MQjava-从零开始实现消息队列-mq02如何实现生产者调用消费者

前景回顾

消费者实现

启动类的调整

MqConsumerHandler 解决类

rpc 音讯体定义

音讯散发

后果回写

调用治理类

接口

实现

TimeoutCheckThread 超时检测线程

音讯生产者实现

启动外围类

MqProducerHandler 生产者解决逻辑

音讯的发送

测试代码

启动消费者

启动生产者

小结

开源地址

拓展浏览