Java小案例-RocketMQ的11种消息类型，你知道几种？（请求应答消息）

news2024/11/28 20:42:55

前言

Rocket的请求应答消息是指在使用Rocket（这里可能是RocketMQ或者Rocket框架）进行通信时，客户端发送一个请求到服务端，然后服务端处理该请求并返回一个响应的过程中的数据交换。

在RocketMQ中：

请求应答消息通常涉及到以下几个步骤：

生产者（Producer）创建一个消息，并将其发送到Broker（消息中间件服务器）。
Broker接收到消息后，可能需要进行存储、路由或者其他处理操作。
如果请求是需要立即响应的（例如RPC调用），Broker会在处理完消息后生成一个响应消息，并通过网络返回给生产者。
生产者接收到响应消息后，可以根据响应内容进行相应的业务处理。

在Rocket框架中：

请求应答消息通常涉及到HTTP请求和响应：

客户端（通常是Web浏览器或者API客户端）向Rocket应用服务器发送一个HTTP请求，请求可能包含JSON、XML或者其他格式的数据。
Rocket框架接收到请求后，根据路由规则将请求分发到对应的处理器函数（handler）。
处理器函数处理请求，这可能包括查询数据库、计算结果或者其他业务逻辑。
处理完成后，处理器函数构建一个HTTP响应，响应中包含处理结果以及可能的状态码、头部信息等。
Rocket框架将响应返回给客户端，客户端解析响应并进行相应的处理。

无论是RocketMQ还是Rocket框架，请求应答消息都是系统间或者组件间通信的基本机制，用于实现功能调用、数据交换或者状态同步。

请求应答消息

这个消息类型比较有意思，类似一种RPC的模式

生产者发送消息之后可以阻塞等待消费者消费这个消息的之后返回的结果

生产者通过过调用request方法发送消息，接收回复消息

public class Producer {
    public static void main(String[] args) throws Exception {
        //创建一个生产者，指定生产者组为 sanyouProducer
        DefaultMQProducer producer = new DefaultMQProducer("sanyouProducer");
        // 指定NameServer的地址
        producer.setNamesrvAddr("192.168.200.143:9876");
        // 启动生产者
        producer.start();


        Message message = new Message("sanyouTopic", "三友的java日记".getBytes());
        
        //发送消息，拿到响应结果， 3000代表超时时间，3s内未拿到响应结果，就超时，会抛出RequestTimeoutException异常
        Message result = producer.request(message, 3000);
        System.out.println("接收到响应消息：" + result);

        // 关闭生产者
        producer.shutdown();
    }

}

而对于消费者来着，当消费完消息之后，也要作为生产者，将响应的消息发送出去

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {

        //创建一个生产者，指定生产者组为 sanyouProducer
        DefaultMQProducer producer = new DefaultMQProducer("sanyouProducer");
        // 指定NameServer的地址
        producer.setNamesrvAddr("192.168.200.143:9876");
        // 启动生产者
        producer.start();


        // 通过push模式消费消息，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sanyouConsumer");
        // 指定NameServer的地址
        consumer.setNamesrvAddr("192.168.200.143:9876");

        // 订阅这个topic下的所有的消息
        consumer.subscribe("sanyouTopic", "*");
        // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("消费消息:%s", new String(msg.getBody()) + "\n");

                    try {
                        // 用RocketMQ自带的工具类创建响应消息
                        Message replyMessage = MessageUtil.createReplyMessage(msg, "这是响应消息内容".getBytes(StandardCharsets.UTF_8));
                        // 将响应消息发送出去，拿到发送结果
                        SendResult replyResult = producer.send(replyMessage, 3000);
                        System.out.println("响应消息的结果 = " + replyResult);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

这种请求-应答消息实现原理也比较简单，如下图所示