前景回顾

上一节我们学习了如何实现基于 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 处理类

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 消息体定义

为了统一标准，我们的 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 唤醒。

但是这也有一个问题，如果一个请求的响应丢失了怎么办？

总不能一直等待吧。

TimeoutCheckThread 超时检测线程

超时检测线程就可以帮我们处理一些超时未返回的结果。

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

MqProducerHandler 生产者处理逻辑

和消费者处理逻辑类似。

这里最核心的就是添加响应结果：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