vlambda博客
学习文章列表

java 从零开始实现消息队列 mq-02-如何实现生产者调用消费者?

前景回顾

小伙伴们好,我是老马。

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

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

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

消费者实现

启动类的调整

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,6060, 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> 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 seqIda70ea2c4325641d6a5b198323228dc24timeoutMills: 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