How to implement springboot rabbitmq reply message direct reply mode

1. Usage Scenarios

The functions of MQ include decoupling, asynchronous, etc.

Usually producers are only responsible for producing messages, and do not care who gets the messages, or what the consumption results are; consumers are only responsible for receiving specified messages for business processing and do not care where the messages come from, first-level reply business processing Condition. But there is a special business in our project. As a message producer, we need to receive the response result of the consumer after producing the message (to put it bluntly, it is similar to the MQ use of synchronous call request response). After research, MQ’s Reply mode (direct reply model) was created for this business model.

2. Reply actual combat

(1) Dependency and YML configuration

Dependency:

I only list the core here Required dependencies for rabbitMq

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Configuration:

No other special configuration, because reply is just an interaction method of rabbitmq

spring:
  rabbitmq:
    host: 10.50.40.116
    port: 5673
    username: admin
    password: admin

(2 ) RabbitMq bean configuration

package com.leilei.demo;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @author lei
 * @create 2022-09-19 21:44
 * @desc mq配置
 **/
@Configuration
public class RabbitMqConfig {
    @Bean
    public Queue bizQueue() {
        return new Queue("bizQueue");
    }
    @Bean
    public Queue replyQueue() {
        return new Queue("replyQueue");
    }
    @Bean
    FanoutExchange bizExchange() {
        return new FanoutExchange("bizExchange");
    }
}

Business class:

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Vehicle implements Serializable {
    private Integer id;
    private String name;
}

(3) Message production end

What the message production end needs to do: Have a message produced , Accept message consumption response

(1) Produce message

• 1. Produce message, depending on the business scenario, choose whether to generate a globally unique custom message Message ID

• 2. Specify the queue for response after message consumption (Reply)

    /**
     * 生产消息
     *
     * @param
     * @return void
     * @author lei
     * @date 2022-09-19 21:59:18
     */
    public void replySend() {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setReplyTo("replyQueue");
        //todo 根据业务，做一个严谨的全局唯一ID,我这里暂时用UUID
        String correlationId = UUID.randomUUID().toString();
        // 我这里指定了唯一消息ID，看业务场景,消费者消费响应后，生产者端可根据消息ID做业务处理
        messageProperties.setCorrelationId(correlationId);
        Vehicle vehicle = new Vehicle(1, "川A0001");
        Message message = new Message(JSON.toJSONString(vehicle).getBytes(), messageProperties);
        rabbitTemplate.convertAndSend("bizExchange","",message);
        System.out.println("生产者发送消息,自定义消息ID为：" + correlationId);
    }

(2) Accept Reply response

After the consumer consumes the message, the processing result will be sent to a queue. By reading the queue here, we can get the response result of the corresponding message for business processing

    /**
     * 接收消息响应
     *
     * @param message
     * @return void
     * @author lei
     * @date 2022-09-19 21:59:27
     */
    @RabbitListener(queues = "replyQueue")
    public void replyResponse(Message message) {
        String s = new String(message.getBody());
        String correlationId = message.getMessageProperties().getCorrelationId();
        System.out.println("收到客户端响应消息ID：" + correlationId);
        //todo 根据消息ID可判断这是哪一个消息的响应,我们就可做业务操作
        System.out.println("收到客户端响应消息：" + s);
    }

(4) Message consumer end

What the message consumer end needs to do is: accept the message and then perform business processing and respond to the message

(1) Method 1: sendTo annotation method return value

Generally speaking, our mq consumer listening method does not need to return a value. If we use the sendTo annotation here, we need to define the message to be responded to as the return value. The sendTo annotation specifies which queue to respond to

Key points:

• 1. The sendTo annotation specifies the corresponding queue (note that it is consistent with the production end)

• 2. Method definition The return value content is the message to be responded to, which will eventually be sent to the corresponding queue specified by the sendTo annotation

• 3. The disadvantage of this method is that the consumer side is very focused, because The target queue specified by sendTo can be written blindly, causing the producer to fail to receive the message response correctly, but I believe that this is not done in general projects

    /**
     * 方式1   SendTo指定响应队列
     *
     * @param message
     * @return String
     * @author lei
     * @date 2022-09-19 16:17:52
     */
    @RabbitListener(queues ="bizQueue")
    @SendTo("replyQueue")
    public String handleEmailMessage(Message message) {
        try {
            String msg=new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("---consumer接收到消息----{}",msg);
            return "客户端响应消息:"+msg+"处理完成!";
        } catch (Exception e) {
            log.error("处理业务消息失败",e);
        }
        return null;
    }

(2) Method 2 : Read the message from the production end and send it using the template

Just like the ordinary consumer method, you only need the RabbitListener annotation to listen to the business queue; but you also need to obtain the ReplyTo address based on the message, and then manually send the message within your own consumer method

• 1. Advantages, you can feel the interactivity of message request and response more strongly, and the process looks clearer

• 2. Disadvantages, code Indecent

    /**
     * 方式2  message消息获取内部reply rabbitmq手动发送
     *
     * @param message
     * @return String
     * @author lei
     * @date 2022-09-19 16:17:52
     */
    @RabbitListener(queues = "bizQueue")
    public void handleEmailMessage2(Message message) {
        try {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("---consumer接收到消息----{}", msg);
            String replyTo = message.getMessageProperties().getReplyTo();
            System.out.println("接收到的reply:" + replyTo);
            rabbitTemplate.convertAndSend(replyTo, "客户端响应消息:" + msg + "处理完成!", x -> {
                x.getMessageProperties().setCorrelationId(message.getMessageProperties().getCorrelationId());
                return x;
            });
        } catch (Exception e) {
            log.error("处理业务消息失败",e);
        }
    }

(3) Method 3: Method return value

This method is actually consistent with 1, but I have tested it because the producer message specifies Without the address of ReplyTo, the consumer does not need to manually specify it again, that is, where to produce the message, whether to respond and where to send the response message are all left to the producer itself. The consumer only needs to process its own business and return results

   /**
     * 方式三  方法有返回值,返回要响应的数据 （reply 由生产者发送消息时指定，消费者不做任何处理）
     *
     * @param message
     * @return String
     * @author lei
     * @date 2022-09-19 23:17:47
     */
    @RabbitListener(queues ="bizQueue")
    public String handleEmailMessage3(Message message) {
        try {
            String msg=new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("---consumer接收到消息----{}",msg);
            return "客户端响应消息:"+msg+"处理完成!";
        }
        catch (Exception e) {
            log.error("处理业务消息失败",e);
        }
        return null;
    }

(4) Test

Production message:

##Consumption message and response:

Response received:

##Link:

The above is the detailed content of How to implement springboot rabbitmq reply message direct reply mode. For more information, please follow other related articles on the PHP Chinese website!