springboot-rabbitmq-reply 消息直接回复模式详情

目录
  • 一、使用场景
  • 二、Reply实战
    • (1)依赖与YML配置
    • (2)RabbitMq bean配置
    • (3)消息生产端
    • (4)消息消费端
      • (1)方法一:sendTo注解+方法返回值
      • (2)方法二:读取生产端的消息使用模板发送
      • (3)方法三:方法返回值
      • (4)测试

一、使用场景

MQ的作用包括了解耦、异步等。

通常生产者只负责生产消息,而不关心消息谁去获取,或者消费结果如何;消费者只负责接收指定的消息进行业务处理而不关心消息从哪里来一级回复业务处理情况。但我们项目中有特殊的业务存在,我们作为消息生产者在生产消息后需要接收消费者的响应结果(说白了就是类似同步调用 请求响应的MQ使用),经过研究,MQ的Reply模式(直接回复模式)就是为此种业务模式而产生。

二、Reply实战

(1)依赖与YML配置

依赖:

我这里只列出最核心的rabbitMq所需依赖

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

配置:

无其余特殊配置,因为reply就是rabbitmq的一种交互方式而已

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

(2)RabbitMq 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、指定消息消费后响应的队列(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)接受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)方法一: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)方法二:读取生产端的消息使用模板发送

与普通的消费者方法一样,只需要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)方法三:方法返回值

这种方式与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)测试

生产消息:

消费消息与响应:

收到的响应:

链路:

如此,MQ版本的请求响应模式就完成了,其实很多大佬使用MQ来实现RPC就是用的ReplyTo啦!

到此这篇关于springboot-rabbitmq-reply 消息直接回复模式详情的文章就介绍到这了,更多相关springboot-rabbitmq-reply 内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

时间: 2022-09-18

SpringBoot整合RabbitMQ实现交换机与队列的绑定

目录 简介 配置方法概述 法1:配置类(简洁方法)(推荐) 法2:配置类(繁琐方法)(不推荐) 法3:使用方配置(不推荐) 法4:MQ服务端网页(不推荐) 简介 本文用实例介绍SpringBoot中RabbitMQ如何绑定交换机(交换器)与队列. 配置方法概述 交换机 下边两种方式等价. ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_WELCOME).durable(true).build(); new TopicExchange(EXCHANGE_T

Springboot整合Rabbitmq之Confirm和Return机制

目录 前言 为什么会有Confirm Springboot整合Mq实现Confirm监听机制 依赖引入 增加配置文件,设定连接信息 配置队列.交换机,以及对其进行绑定 编写mq消息发送服务 编写消息发送接口 启动项目进行测试 正常测试 异常测试 什么是Return? 增加ReturnCallback监听并测试 修改RabbitmqService配置类 测试 总结 相关代码下载 前言 之前专栏中,对Springboot整合Rabbitmq都有一系列的配置和说明,但总缺少一些必要的描述信息.导致很多

SpringBoot 使用log4j2的配置过程

目录 前言 日志接口(slf4j) 日志实现(log4j.logback.log4j2) 配置过程 涉及的POM部分文件 前言 日志接口(slf4j) slf4j是对所有日志框架制定的一种规范.标准.接口,并不是一个框架的具体的实现,因为接口并不能独立使用,需要和具体的日志框架实现配合使用(如log4j.logback). 接口用于定制规范,可以有多个实现,使用时是面向接口的(导入的包都是slf4j的包而不是具体某个日志框架中的包),即直接和接口交互,不直接使用实现,所以可以任意的更换实现而不用

Springboot整合RabbitMq测试TTL的方法详解

目录 什么是TTL? 如何设置TTL? 设定整个队列的过期时间 配置类编写 测试 配置 测试 总结 代码下载 什么是TTL? 在RabbitMq中,存在一种高级特性 TTL. TTL即Time To Live的缩写,含义为存活时间或者过期时间.即: 设定消息在队列中存活的时间.当指定时间内,消息依旧未被消费,则由队列自动将其删除. 如何设置TTL? 既然涉及到设定消息的存活时间,在RabbitMq中,存在两种设置方式: 设置整个队列的过期时间. 设置单个消息的过期时间. 设定整个队列的过期时间

springboot+rabbitmq实现智能家居实例详解

目录 引言 一.什么是 MQTT协议? 二.为什么要用 MQTT协议? 三.MQTT协议介绍 MQTT数据包 1.固定头 2.可变头 3.消息体payload 消息质量(QoS ) 1.Qos 0 2.Qos 1 3.Qos 2 LWT(最后遗嘱) 四.MQTT协议应用场景 五.代码实现 1.启用 rabbitmq的mqtt协议 2.mqtt 客户端依赖包 3.消息发送者 4.消息订阅 六.测试消息 1.测试消息发送 2.测试消息订阅 七.应用注意事项 clientId 要唯一 八.其他中间件

详解SpringBoot整合RabbitMQ如何实现消息确认

目录 简介 生产者消息确认 介绍 流程 配置 ConfirmCallback ReturnCallback 注册ConfirmCallback和ReturnCallback 消费者消息确认 介绍 手动确认三种方式 简介 本文介绍SpringBoot整合RabbitMQ如何进行消息的确认. 生产者消息确认 介绍 发送消息确认:用来确认消息从 producer发送到 broker 然后broker 的 exchange 到 queue过程中,消息是否成功投递. 如果消息和队列是可持久化的,那么确认消

SpringBoot+RabbitMQ实现消息可靠传输详解

目录 环境配置 消息丢失分析 生产阶段 生产端模拟消息丢失 RabbitMQ 消费端 环境配置 SpringBoot 整合 RabbitMQ 实现消息的发送. 1.添加 maven 依赖 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <depen

SpringBoot利用AOP实现一个日志管理详解

目录 1. 需求 2. 新建一张日志表 3. 写相应的Controller层 4.Service接口层 5.Service实现 6.Mapper接口 7.Mapper.xml(我用的是Mybatis) 8.CspLog 9.实体类SysOperCspLog 10. 定义日志管理的切面 11.AsyncFactoryCsp 12. 写一个Controller的Demo来执行一条日志试试 1. 需求 目前有这么个问题,有两个系统CSP和OMS,这俩系统共用的是同一套日志操作:Log;目前想区分下这俩

MySQL日志管理详解

日志文件对于一个服务器来说是非常重要的,它记录着服务器的运行信息,许多操作都会写日到日志文件,通过日志文件可以监视服务器的运行状态及查看服务器的性能,还能对服务器进行排错与故障处理,MySQl中有六种不同类型的日志. 一.日志种类 ---–> 1,错误日志:记录启动.运行或停止时出现的问题,一般也会记录警告信息. ---–> 2,一般查询日志:记录建立的客户端连接和执行的语句. ---–> 3,慢查询日志:记录所有执行时间超过long_query_time秒的所有查询或不使用索引的查询,

Node.js log4js日志管理详解

前言 在Nodejs中使用express框架并没有自带的日志模块,我们可以选择log4js来完成日志记录的功能. 如果用过JAVA中log4j的同学,肯定对日志并不陌生,学习log4js会更得心应手的. 项目要用log4js记录日志,网上找的教程都是很久以前的,新版本log4js跟旧版本有一些不同,看文档结合旧版本配置了能用的log4js,记录一下,当前log4js版本2.3.12,采用按天分片的方式,同时输出到控制台和文件. log4js单独配置到一个文件,其他要使用的引用这个文件. log-

Swift利用CoreData实现一个通讯录存储详解

前言 相信大家对CoreData应该都不陌生,CoreData作为Apple的亲儿子,依然在App需要存储结构化数据上发挥着重要的作用.CoreData已经超过十年了,而且亲爹还在积极的维护着它. Core Data是iOS5之后才出现的一个数据持久化存储框架,它提供了对象-关系映射(ORM)的功能,即能够将对象转化成数据,也能够将保存在数据库中的数据还原成对象. 虽然其底层也是由类似于SQL的技术来实现,但我们不需要编写任何SQL语句,有点像Java开发中的Hibernate持久化框架 Cor

springboot配置logback日志管理过程详解

这篇文章主要介绍了springboot配置logback日志管理过程详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 首先书写logback-spring.xml文件为: <?xml version="1.0" encoding="UTF-8"?> <configuration> <springProperty scope="context" name="

springboot使用logback文件查看错误日志过程详解

这篇文章主要介绍了springboot使用logback文件查看错误日志过程详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 <?xml version="1.0" encoding="UTF-8"?> <!-- 从高到地低 OFF . FATAL . ERROR . WARN . INFO . DEBUG . TRACE . ALL --> <!-- 日志输出规则 根据当前ROOT

springboot config 拦截器使用方法实例详解

本文介绍Spring-Boot中使用拦截器,一般在拦截器中处理跨域处理,允许跨域访问项目,拦截器使用详细资料请查阅官网. 实现自定义拦截器步骤: 1.创建一个类并实现HandlerInterceptor接口. 2.创建一个Java类继承WebMvcConfigurerAdapter,并重写 addInterceptors 方法. 2.将自定义的拦截器交由spring管理,然后将对像手动添加到拦截器链中(在addInterceptors方法中添加). 创建拦截器类 package com.exam

利用二进制文件安装etcd的教程详解

etcd组件作为一个高可用强一致性的服务发现存储仓库. etcd作为一个受到ZooKeeper与doozer启发而催生的项目,除了拥有与之类似的功能外,更专注于以下四点. 简单:基于HTTP+JSON的API让你用curl就可以轻松使用. 安全:可选SSL客户认证机制. 快速:每个实例每秒支持一千次写操作. 可信:使用Raft算法充分实现了分布式. 场景一:服务发现(Service Discovery)一个强一致性.高可用的服务存储目录.基于Raft算法的etcd天生就是这样一个强一致性高可用的

Crashlytics Android 异常报告统计管理(详解)

简介 Crashlytic 成立于2011年,是专门为移动应用开者发提供的保存和分析应用崩溃信息的工具.Crashlytics的使用者包括:支付工具Paypal, 点评应用Yelp, 照片分享应用Path, 团购应用GroupOn等移动应用. 2013年1月,Crashlytics被Twitter收购,成为又一个成功的创业产品.被收购之后,由于没有了创业公司的不稳定因素,我们更有理由使用它来分析应用崩溃信息. 使用Crashlytics的好处有: 1.Crashlytics不会漏掉任何应用崩溃信

springboot中使用redis的方法代码详解

特别说明: 本文针对的是新版 spring boot 2.1.3,其 spring data 依赖为 spring-boot-starter-data-redis,且其默认连接池为 lettuce ​redis 作为一个高性能的内存数据库,如果不会用就太落伍了,之前在 node.js 中用过 redis,本篇记录如何将 redis 集成到 spring boot 中.提供 redis 操作类,和注解使用 redis 两种方式.主要内容如下: •docker 安装 redis •springboo

SpringBoot集成WebSocket长连接实际应用详解

前言: 一.WebSocket之初出茅驴 官方定义:WebSocket是一种在单个TCP连接上进行全双工通信的协议.WebSocket使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据.在WebSocket API中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输.是真正的双向平等对话,属于服务器推送技术的一种. 太官方啦,还是博主过来翻译一下吧 :WebSocket技术只需要service和client建立一次连接,就能实现服