springboot Rabbitmq 応答メッセージの直接応答モードを実装する方法-＆＃＆チュートリアル-php.cn

springboot Rabbitmq 応答メッセージの直接応答モードを実装する方法

王林

リリース： 2023-05-11 11:58:14

転載

1410 人が閲覧しました

1. 使用シナリオ

MQ の機能には、デカップリング、非同期などが含まれます。

通常、プロデューサーはメッセージを作成することのみを担当し、誰がメッセージを取得するか、または消費結果が何であるかは気にしません。コンシューマーは、ビジネス処理のために指定されたメッセージを受信することのみを担当し、メッセージがどこに送信されるかは気にしません。 from、第 1 レベルの応答ビジネス処理条件。しかし、私たちのプロジェクトには特別なビジネスがあり、メッセージプロデューサーとして、メッセージを生成した後にコンシューマーの応答結果を受け取る必要があります (率直に言うと、これは同期呼び出し要求応答を MQ で使用するのと似ています)。調査によると、MQ の返信モード (直接返信モデル) は、このビジネスモデル用に作成されました。

2. 実際の戦闘への返信

(1) 依存関係と YML 構成

Dependency:

ここではコアのみをリストします。 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 ) の対話メソッドであるため、その他の特別な構成はありません。 Bean 設定

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) メッセージ生成側

メッセージ生成側で行うべきこと: メッセージを生成する、メッセージ消費応答の受け入れ

(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) Accept Reply 応答

コンシューマーがメッセージを消費した後、処理結果がキューに送信されます。ここでキューを読み取ることで、業務処理用の該当メッセージの応答結果を取得できます

    /**
     * 接收消息响应
     *
     * @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 アノテーションメソッドの戻り値

通常、mq コンシューマリスニングメソッドは値を返す必要はありませんが、ここで sendTo アノテーションを使用する場合は、戻り値として応答するメッセージを定義する必要があります。 sendTo アノテーションは、どのキューに応答するかを指定します。

重要なポイント:

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

ログイン後にコピー

このメソッドは実際には 1 と一致しますが、プロデューサーのメッセージが ReplyTo のアドレスなしで指定しているため、テストしました。コンシューマは、それを手動で再度指定する必要はありません。つまり、メッセージを生成する場所、応答するかどうか、および応答メッセージをどこに送信するかはすべて、プロデューサ自体に任されています。コンシューマは、独自のビジネスを処理して結果を返すだけで済みます。

   /**
     * 方式三  方法有返回值,返回要响应的数据 （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) テスト

生産メッセージ:

springboot rabbitmq reply消息直接回复模式怎么实现 #消費メッセージと応答: