springboot Rabbitmq 응답 메시지 직접 응답 모드를 구현하는 방법-java지도 시간-php.cn

springboot Rabbitmq 응답 메시지 직접 응답 모드를 구현하는 방법

王林

풀어 주다： 2023-05-11 11:58:14

앞으로

1474명이 탐색했습니다.

1. 사용 시나리오

MQ의 기능에는 디커플링, 비동기 등이 포함됩니다.

일반적으로 생산자는 메시지 생성에만 책임이 있으며 누가 메시지를 받는지 또는 소비 결과가 무엇인지는 신경 쓰지 않습니다. 소비자는 비즈니스 처리를 위해 지정된 메시지를 받는 것만 책임지며 메시지가 어디서 왔는지 비즈니스에 응답하는지 신경 쓰지 않습니다. 처리 상태. 하지만 우리 프로젝트에는 특별한 비즈니스가 있습니다. 메시지 생성자로서 우리는 메시지를 생성한 후 소비자의 응답 결과를 받아야 합니다(직접 말하면 MQ에서 동기 호출 요청 응답을 사용하는 것과 비슷합니다). 연구를 통해 MQ의 응답 모드(직접 응답 모델)가 이 비즈니스 모델을 위해 만들어졌습니다.

2. 응답 실행

(1) 종속성 및 YML 구성

종속성:

rabbitMq

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

로그인 후 복사

에 필요한 핵심 종속성만 나열합니다.

다른 특별한 구성은 없습니다. RabbitMQ의 대화형 방식

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

로그인 후 복사

(2) RabbitMq 빈 구성

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

로그인 후 복사

비즈니스 클래스:

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

로그인 후 복사

(3) 메시지 생성 종료

메시지 생성 종료에서 해야 할 일: 메시지 생성 및 수락 Message 소비 응답

(1) 메시지 생성

1. 비즈니스 시나리오에 따라 메시지 생성, 전역적으로 고유한 사용자 지정 메시지 ID
2를 생성할지 여부를 선택합니다. (답장)

    /**
     * 生产消息
     *
     * @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) 응답 응답 수락

소비자가 메시지를 소비한 후 처리 결과가 대기열로 전송됩니다. 이 대기열을 읽으면 해당 응답 결과를 얻을 수 있습니다. 비즈니스 처리를 위한 메시지

    /**
     * 接收消息响应
     *
     * @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) 메시지 소비자 측

메시지 소비자가 해야 할 일은: 메시지를 수락한 다음 비즈니스 처리를 수행하고 메시지에 응답하는 것입니다.

(1) 방법 1: sendTo 주석 + 메서드 반환 value

일반적으로 우리는 mq를 사용합니다. 리스너 메서드에는 반환 값이 필요하지 않습니다. 여기서는 sendTo 주석을 사용하며, 응답할 메시지는 반환 값으로 정의되어야 합니다. sendTo 주석은 응답할 대기열을 지정합니다. to

핵심 사항:

1. sendTo 주석은 해당 대기열을 지정합니다(생산 종료와 일치함)
2 메서드에서 정의한 반환 값 내용은 메시지입니다. 이는 결국 해당 대기열을 지정하기 위해 sendTo 주석으로 전송됩니다
3. 이 방법의 단점은 sendTo로 지정된 대상 대기열이 맹목적으로 작성될 수 있기 때문에 소비자 측이 매우 중요하다는 것입니다. 생산자 측에서 메시지 응답을 제대로 받지 못하는 현상이 발생하는데, 일반 프로젝트에서는 그렇지 않은 것 같아요

    /**
     * 方式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) 방법 2: 제작 측에서 메시지를 읽어보고 템플릿을 이용해 보내주세요

일반 소비자 메서드와 마찬가지로 비즈니스 대기열을 수신하려면 RabbitListener 주석만 필요하지만 메시지를 기반으로 ReplyTo 주소를 얻은 다음 자체 소비자 메서드 내에서 수동으로 메시지를 보내야 합니다.

1. 장점, 메시지 요청과 응답의 상호작용성을 더 강하게 느낄 수 있고, 프로세스가 더 명확해 보입니다
2. 단점, 코드가 세련되지 않습니다

    /**
     * 方式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) 방법 3: 메서드 반환 값

이 방법은 실제로는 1과 일치하지만, 생산자 메시지가 ReplyTo의 주소를 지정하기 때문에 소비자가 수동으로 메시지를 생성할 위치, 응답할지 여부, 위치를 다시 지정할 필요가 없습니다. 응답 메시지를 보내려면 생산 종료 자체가 비어 있고 소비자는 자체 비즈니스를 처리하고 결과만 반환하면 됩니다. 응답:

받은 응답:

springboot rabbitmq reply消息直接回复模式怎么实现