快速搞定分布式RabbitMQ---RabbitMQ进阶与实战

news2024/11/23 3:27:53

本篇内容是本人精心整理；主要讲述RabbitMQ的核心特性；RabbitMQ的环境搭建与控制台的详解；RabbitMQ的核心API；RabbitMQ的高级特性;RabbitMQ集群的搭建；还会做RabbitMQ和Springboot的整合；内容会比较多，将原理跟应用相结合，希望大家能有所收获！

1.初始RabbitMQ核心概念

RabbitMQ是一个开源的消息代理与队列服务器，用来通过普通协议在完全不同
的应用之间共享数据，RabbitMQ是使用Erlang语言来编写的，并且RabbitMQ
是基于AMQP协议的。
MQ使用的场景：
电商系统跟物流体系：体系之间(为什么可以使用MQ，
因为对于消息的传递无非就是几种模式，rpc通信(包括http，主流的java框架
Springcloud，springboot都是；)，)
逻辑：即时性很强，比如付款马上知道是否成功，一般来讲都会采用rpc
或者http请求
服务解耦；削峰填谷；异步化；

本身也可完成消息堆积的这件事情。消息堆积能力有限。
解决方式：要么有足够的consumer;消费处理能力比较快

语言分析：
Erlang语言：
一种交换机语言，强大点在于：数据同步非常快(节点与节点之间)
**AMQP协议：**高级消息队列协议；
定义：是具有现代特征的二进制协议，是一个提供统一消息服务的
应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向
消息的中间件设计。
AMQP核心概念：
Server:又称Broker，接受客户端的连接，实现AMQP实体服务
Connection:连接，应用程序与broker的网络连接
Channel:网络信道，几乎所有的操作都在channel中进行，Channel是
进行消息读写的通道。客户端可以建立多个channel，每个channel
代表一个会话任务。
Message:消息，服务器与应用程序之间传送的数据，由Properties和Body组成.Properties可以对消息进行修饰，必须消息的优先级、延迟等高级特性；Body则是消息体内容。

协议模型：
生产者产生通过rabbitMQ去把消息投递到rabbitMQ，
通过复杂的exchange放入Message Queue中；消费者从messageQueue
中获取消息
在这里插入图片描述
server：代表rabbitMQ
virtualhost: 虚拟主机；可以进行作用域的划分；虚拟地址，用于进行逻辑隔离，最上层的消息路由。
一个virtual host里面可以有若干个Exchange和Queue，同一个Virtual
Host 里面不能有相同名称的Exchange 或 Queue。
Exchange：交换机，接收消息，根据路由键转单消息到绑定队列(主题。与message queue相关，决定数据发送到那个message)
Binding: Exchange和Queue之间的虚拟链接，binding中可以包换routing key
Routing key: 一个路由规则，虚拟机可用它来确定如何路由一个特定消息。（如负载均衡）
Queue:消息队列。多对多的关系，应用场景下多为一对多的关系).

Rabbit的整理架构：
实际工作中一个consumer消费一个queue最好。
应用场景下，一个exchange对多个queue;

在这里插入图片描述
消息是如何流转的：
在这里插入图片描述

2.RabbltMQ环境搭建与控制详解

(1)RabbltMQ安装教程：
这里使用RabbitMQ3.6.5版本进行操作
环境搭建：
官网地址：http://www.rabbitmq.com/
环境描述：linux (centos 7 RedHat7)
1.首先在linux上进行一些软件的准备工作，yum下来一些基础的软件包
注：总共三个节点需要配置
yum install build-essential openssl openssl-devel unixOOBC-idevel make gcc
gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

配置好主机名称
/etc/hosts /etc/hostname

2.下载RabbitMQ所需软件包(这里使用的是RabbitMQ 3.6.5稳定版本)

3.安装服务命令：
rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm

rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm(命令报错)
错误信息：依赖检测失败：tcp_wrappers 被 socat-1.7.3.2-1.1.el7.x86_64 需要
使用 rpm -ivh socat-1.7.3.2-1.1.el7.x86_64.rpm --force --nodeps

rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm

4.修改用户登录与连接心跳检测，注意修改
vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app
修改点1:loopback_users 中的<<”guest”>>,只保留guest(用于用户登录)
(修改前：{loopback_users, [<<“guest”>>]}, 这里只需要将<<>>删除即可)
修改点2：heartbeat 为10(用于心跳连接)

5.安装控制插件

5.1首先启动服务(后面|包含了停止，查看状态以及重启的命令)
/etc/init.d/rabbitmq-server start | stop | status |restart

5.2 查看服务有没有启动：lsof -i:5672(5672是rabbit的默认端口)
rabbitmq-plugins enable rabbitmq_management

5.3可查看管理端口有没有启动:
lsof -i:15672 或者netstat -tnlp| grep 15672
6. 一切ok，我们访问地址，输入用户名密码均为 guest:
http://你的ip地址：15672/

总结：
1.防火墙
需要添加5672以及15672端口，防火墙允许其开放
第一个命令：firewall-cmd --add-port=5672/tcp --permanent
显示执行成功：success
但是使用：firewall-cmd --query-port=5672/tcp查询的时候未查询到
2.hostname
修改涉及集群的测试；Hostname,修改后需要重启才能够生效；

(2)rabbitMQ控制台相关的内容:
rabbitMQ自带的交换机:
在这里插入图片描述
看不到相应的文件在哪，可以查看控制台：
在这里插入图片描述
安装文档中，配置文件的地址也是有的：这两个是有优先级的，
默认是控制台的优先级比较高
/usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app

测试：
在这里插入图片描述
1.创建exchange：
exchange的type实际工作中使用的比较多的是direct(直连)和topic(主题)。
fanout是广播.
Durability:持久化设置durable：持久化，transient：不持久化。
不持久化的时候，服务重启exchange就会消失。
arguments:表示参数

创建队列：
在这里插入图片描述
如何将exchange与queue建立关系：
可以在控制台选择exchange操作，也可以在控制台的queue中选择对应的
queue添加exchange就可。

绑定成功：
在这里插入图片描述
发布消息：
使用特定routing key消息才能够发送过来。
通过binding实现数据发送
使用queue对应的routing key，数据发送成功。
在这里插入图片描述
此时，数据显示为两条。
在这里插入图片描述
当不使用对应routing key：
在这里插入图片描述
Butnot routed:表示发布失败。
在这里插入图片描述
数据的总数没变。

3.RabbitMQ急速入门Helloworld

ConnectionFactory:获取连接工厂
Connection：一个连接
Channel:数据通信信道，可发送和接收消息
Queue:具体消息存储队列
Prodecer&Consumer:生产和消费者

核心：

<dependency>
   <groupId>com.rabbitmq</groupId>
   <artifactId>amqp-client</artifactId>
   <version>3.6.5</version>
</dependency>

Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来
发送和接收消息，包含两部分，其中spring-amqp是基础抽象，
Spring-rabbit是底层的默认实现。
SpringAMQP提供了三个功能：
自动声明队列、交换机及其绑定关系
基于注解的监听器模式，异步接收消息
封装了RabbitTemplate工具，用于发送消息

生产者Sender：

ConnectionFactory connectionFactory  = new ConnectionFactory();
connectionFactory.setHost("192.168.56.107");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
//2.创建ConnectionFactory
Connection connection = connectionFactory.newConnection();
//3.创建channel
Channel channel = connection.createChannel();
// 4 声明
String queueName = "testone";
///  参数: queue名字,是否持久化,独占的queue（仅供此连接）,不使用时
是否自动删除, 其他参数
channel.queueDeclare(queueName,false,false,false,null);
Map<String, Object> headers = new HashMap<>();
//deliveryMode 1:消息非持久化，2：消息持久化
AMQP.BasicProperties props = new AMQP.BasicProperties().builder()
        .deliveryMode(2)
        .contentEncoding("UTF-8")
        .headers(headers).build();
for(int i = 0; i < 5;i++) {
    String msg = "Hello World RabbitMQ " + i;
    //channel第一没有设置exchange的名称，是按照queueName来发送消息
    //使用的是默认的exchange，props就是上面的props。
    channel.basicPublish("", queueName , props , msg.getBytes());
}

备注：队列持久化与消息持久化是区分开的。
只有队列跟消息都持久化时，重启之后消息才能够保存。

是否自动ack：ack就是确认收到消息发送的一个ack确认。
Offset：下一条需要处理的消息的编号。
在这里插入图片描述
补充：同步的情况下，没有ack；只有在异步的时候，才能回发ACK，ack会将offset传回去。

正常的生产环境，一般使用的是手动ack，因为不确定消息是否成功的传输，所以一定要手动ack
消费者:Receiver

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setHost("192.168.56.107");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

//服务如果宕机的话，可以进行自动切换。
connectionFactory.setAutomaticRecoveryEnabled(true);
//设置网络恢复间隔
connectionFactory.setNetworkRecoveryInterval(3000);
Connection connection = connectionFactory.newConnection();

Channel channel = connection.createChannel();

String queueName = "testone";
// durable是否持久化消息
channel.queueDeclare(queueName, false, false, false, null);
QueueingConsumer consumer = new QueueingConsumer(channel);
//参数：队列名称，是否自动ACK，Consumer,相当于设置监听
channel.basicConsume(queueName,true,consumer);
//循环获取消息

while (true){
    //获取消息，如果没有消息，这一步将会一直阻塞
    Delivery delivery = consumer.nextDelivery();
    String msg = new String(delivery.getBody());
    System.out.println("收到消息"+msg);
 }

4.RabbltMQ核心API

(1)API-exchange之Direct
Exchange：发送和接收消息；并根据路由键转发消息
所绑定的队列
在这里插入图片描述
消息是根据你的路由key加上你的exchange所对应的
决定他到底传到那个队列；

交换机属性：
Name:交换机名称；
Type:交换机类型 direct，topic，fanout，headers
Durability:是否需要持久化，true为持久化
AutoDelete:当最后一个绑定到Exchange上的队列删除后，
自动删除该exchange(一般设置不自动)；
Internal:当前exchange是否用于RabbitMQ内部使用，默认为False
Arguments：扩展参数，用于扩展AMQP协议自制定义使用，
在这里插入图片描述
在这里插入图片描述
首先设置exchange的name，然后设置queue信息。
最后通过queueBind实现三者的binding关系，生产者
只关注发到那个exchange和携带的routingKey,
direct模式要求exchangename必须与routingkey完全匹配，不能模糊匹配。
在这里插入图片描述
当routingkey和队列的名字一样，也能发过去；
走的是amqp.defaultexchanage

(2)API-exchange之Topic
所有发送的TopicExchange 的消息都被转发到所有关心RouteKey
中指定的Topic的Queue上。
Exchange将RouteKey和某topic进行模糊匹配，此时队列需要
绑定一个topic
注：可以使用通配符进行模糊匹配
在这里插入图片描述
在这里插入图片描述
最好别用上面的匹配规则，建议一种消息就搞一种匹配规则就好了。

问题：
consumer2为什么也能够消费user.delete.abc.
在这里插入图片描述

因为对于统一个queue绑定了两个不同的规则同一个队列，绑定了两个规则，所以这两个消费者无论谁都可以随机消费，
但如果两个规则对应两个不同的队列，两个消费者
也消费各自的队列，这时候就按匹配规则来了，

另外，如果两个消费者监听同一个队列，那么他俩会均摊消息不回重复消费，但如果监听各自不同的队列，而且发送方也都能发送给这两个队列，消费者也消费两个不同的队列，那这个时候就各自消费各自的了

备注：现实情况，最好使用控制台去创建台，去绑定key，queue
以及对应的exchange之间的关系。

(3)API-exchange之Fanout
不处理路由键，只需要简单的将队列绑定到交换机上。
发送到交换机的消息会被转发到与该 交换机绑定的所有队列上
Fanout交换机转发消息是最快的。
在这里插入图片描述
(4)API-其他关键概念讲解
Binding-绑定
Exchange和exchange，queue之间的连接关系；
Binding可以包含RoutingKey和参数
队列：Queue

Message:
服务器和应用程序之间传送的数据
本质上就是一段数据，由properties和Payload(Body)组成
常用属性：delivery mode.headers(自定义属性)；
自定义属性：content_type,content_encoding,piority;
correlation_id,reply_to,expiration,message_id;
Timestamp,type,user_id,app_id,cluster_id;
备注：correlation_id可以作为唯一标记，message_id也是可以当成
唯一的id.expiration:可以用来设置过期时间。
Virtual host:虚拟主机
虚拟地址：用于进行逻辑隔离，最上层的消息路由；
一个virtual host里面可以有若干个Exchange和Queue;
同一个Virtual host里面不能有相同的名称和Exchange和Queue。

5.RabbitMQ高级特性

(1)生产端可靠性投递与消费端幂等性
需要解决的问题：
如何保障消息100%的投递成功？
幂等性概念详解
在海量订单产生的业务高峰期，如何避免消息的重复消费问题？
Confirm确认消息，return返回消息；
消息的ACK与重回队列
消息的限流；
TTL消息；生存时间
死信队列；

如何保障消息100%的投递成功？
什么是生产端的可靠性投递？
保障消息的成功发出；保障MQ节点成功接收；
发送端收到MQ节点(Broker)确认应答；完善的消息补偿机制；

BAT/TMD互联网大厂的解决方案：
机制一：消息落库，对消息的状态进行打标；机制二：分两次发送
在这里插入图片描述
BIZDB:就是自己的业务，MSGDB一个记录(理解为log日志记录)。
理解：
业务入库要做一个log日志。(同源：一个链接可以去操作
一个地址下面多个数据库，一个connection，它在一个事务内的)
Step1：业务入库；step2：日志记录；(上面两个step需要实现原子性)
Step3：将消息发送到MQ Broker上。(需要broker返回ack，确认消息发送成功)
Step4:传递ack；
Producer Component:监听，接受broker端给与的ACK，
Step5：更新MSGDB,更新状态，表示发送成功。
(当一直没有ack，这时你可以设置超时时间，用定时任务去扫描日志表
里状态一直是0的，然后再将对应的数据进行重发。)

幂等性概念：
类比数据库的乐观锁机制；sql语句
在这里插入图片描述
在海量订单产生的业务高峰期，如何避免消息的重复消费问题：
消费端实现幂等性，意味着，我们的消息永远不会消费多次；
即使我们收到了多条一样的消息；

业界主流的幂等性操作：
业务唯一ID或者指纹码机制，利用数据库主键去重
在这里插入图片描述
好处：实现简单

(2)生产端特性讲解-确认机制和返回机制
理解Confirm消息确认机制：
消息的确认，是指生产者投递消息后，如果Broker收到消息，
则会给生产者一个应答；
生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，
进行这种方式也是消息的可靠性投递的核心保障

Confirm消息的流程：brokerconfirm就是ack。
在这里插入图片描述
在这里插入图片描述
Return消息机制：
ReturnListener:用于处理一些不可路由的消息；
我们消息生产者，通过制定一个exchange和Routingkey,
把消息送达到某队列中，然后消费者监听队列，进行消费处理；
但是在某些情况下，如果我们再发送消息的时候，当时的exchange不存在
或者执行的路由key路由不到，这个时候我们需要监听这种不可达的消息
就需要return Listener!
基础api中有一个关键的配置项；
Mandatory:如果为true，则监听器会收到路由不可达的消息，然后进行
后续处理；如果为false，则，broker端会自动删除该消息。
流程：
Producer发送消息；路由不可达或者exchange不存在的时候；
broker会丢弃掉这条消息；producer要知道消息发送情况，这
就存在Return Listener机制。

代码示例：
Confirm(Sender)核心部分：

//采用confirm模式，异步实现
channel.confirmSelect();
//添加监听，等待broker应答
channel.addConfirmListener(new ConfirmListener() {
   @Override
   public void handleNack(long deliveryTag, boolean multiple) throws IOException {
      System.err.println("------- error ---------");
   }
   @Override
   public void handleAck(long deliveryTag, boolean multiple) throws IOException {
      System.err.println("------- ok ---------");
   }
});
      
channel.basicPublish(exchangeName, routingKey1 , null , msg.getBytes());

不能保证百分之百发送成功，因为可能消息发到broker之后，
Broker发送的ack没收到，网络中断了。压根没收到回调函数。就需要补偿机制。

Return在满足对应机制时是没有回应的。不调用handleReturn
Return(Sender)核心部分：

//5 监听
channel.addReturnListener(new ReturnListener() {
public void handleReturn(int replyCode,
                        String replyText,
                        String exchange,
                        String routingKey,
                        AMQP.BasicProperties properties,
                        byte[] body)
         throws IOException {
      System.out.println("**************handleReturn**********");
      System.out.println("replyCode: " + replyCode);
      System.out.println("replyText: " + replyText);
      System.out.println("exchange: " + exchange);
      System.out.println("routingKey: " + routingKey);
      System.out.println("body: " + new String(body));
   }
   });

(3)流控服务和ACK重回队列
消费端限流：
假设有个场景：首先，我们rabbitMQ服务器有上万条未处理的消息
我们随便打开一个消费者客户端，会出现下面的情况：

巨量的消息全部推送过来，但是我们单个客户端无法同时处理这么
多数据！
RabbitMQ提供了一种qos(服务质量保证)，即在非自动确认消息的
前提下，如果一定数目的消息(通过基于consume或者channel设置
Qos的值)未被确认前，不进行消费消息。

Void BasicQos(unit prefetchSize ,unshort prefetchCount,bool global);

参数：prefetchSize：报文大小
prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于
N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，
知道有消息ack
Global：true/false:是否将上面设置应用于channel。
**理解：**global就是上面限制是channel级别还是consumer级别
在这里插入图片描述
消费端的手工ACK与NACK
消费端进行消费的时候，如果由于业务异常，我们可以进行日志标记，
然后进行补偿(尽量不要做重回队列)
如果由于服务器宕机等严重问题，那我们就需要手工进行ACK，
保障消费端消费成功

消息的重回队列：
消息的重回队列是为了对没有处理成功的消息，把消息重新递给Broker!
在实际工作中，都会关闭重回队列，设置为false(避免多次重回)

代码示例：
核心部分(Receiver):

//  参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, false, consumer);  
// 循环获取消息  
while(true){  
    // 获取消息，如果没有消息，这一步将会一直阻塞  
    Delivery delivery = consumer.nextDelivery();  
    String msg = new String(delivery.getBody());    
    System.out.println("收到消息：" + msg);  
    Thread.sleep(1000);
    
    if((Integer)delivery.getProperties().getHeaders().get("flag") == 0) {
       //throw new RuntimeException("异常");
       channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
    } else {
       channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

在这里，nack时候可以使用重回队列(与ack的主要区别)。
设置重回队列的话，一直会打印消息，因为一直在重回队列。

关于流控的内容：如下示例。
在这里插入图片描述
在传递消息前添加basicQos设置就可。

(4)TTL消息与死信队列详解

TTL队列/消息：(生存时间，Time to Live)
RabbitMQ支持消息的过期时间，在消息发送时可以进行指定
RabbiitMQ支持队列的过期时间，从消息入队列开始计算，只要超过了
队列的超时时间配置，那么消息会自动删除。

**死信队列：**DLX,Dead-Letter-Exchange.
利用DLX，当消息在某一个队列中变成死信(dead message)之后，
它能被重新publish到另一个exchange，这个exchange就是DLX.

消息变成死信有几种情况：
消息被拒绝(basic.reject/basic.nack),并且requeue=false(重回队列=false);
消息的TTL过期；队列达到最大长度；
DLX是一个正常的exchange，和一般的Exchange没有区别，它能在任何队列
上被指定，实际上就是设置某个队列的属性。
当某个队列中有死信时，rabbitMQ会自动的将这个消息发布到这只的exchange上去。
进而被路由到另一个队列。
可以监听这个队列中消息做相应处理，这个特性弥补rabbitMq3.0
以前支持的immediate参数的功能。

死信队列的设置：
首先，需要设置死信队列的exchange和queue，然后进行绑定。
Exchange:dlx.exchange; Queue:dlx.queue;RoutingKey:#;(自己定义)
然后我们进行正常声明交换机，队列，绑定，只不过我们需要在队列加上
一个参数即可。Arguments.put(“x-dead-letter-exchange”,dlx.exchange);
这样消息在过期，requeue,队列在达到最大长度时，消息就可以直接
路由到死信队列！

代码示例：dlx

//1 创建ConnectionFactory
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.56.107");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

//2 创建Connection
Connection connection = connectionFactory.newConnection();
//3 创建Channel
Channel channel = connection.createChannel();  
//4 声明
String exchangeName = "test_dlx_exchange";
String routingKey = "group.bfxy";
//5 发送

Map<String, Object> headers = new HashMap<String, Object>();

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.contentEncoding("UTF-8")
// TTL
.expiration("6000")
.headers(headers).build();

String msg = "Hello World RabbitMQ 4 DLX Exchange Message ... ";
channel.basicPublish(exchangeName, routingKey , props , msg.getBytes());

6.RabbitMQ集群搭建-镜像队列集群搭建环境搭建实操

镜像模式：
集群模式非常经典的就是Mirror镜像模式，保证数据100%
不丢失，在实际工作中也是用的最多的，并且实现集群非常简单。
一般互联网大厂都会构建这种镜像集群模式。
Mirror镜像队列，目的是为了保证rabbitmq数据的高可靠性的解决方案。
主要是实现数据的同步，一般来讲是2-3个节点实现数据同步
(对于100%数据可靠性解决方案一般是3个节点)集群架构如下：
在这里插入图片描述
具体安装步骤参考安装文档:
Haproxy+keepalived可以参考我之前写的文章，执行安装！
参考地址：
https://blog.csdn.net/TOMORROW6COME/article/details/140296869
节点参考：分配节点，定义安装集群；
修改hostname即可。

安装成功：
在这里插入图片描述
在这里插入图片描述
镜像模式配置成功：+2表示两个节点。可以在控制台看到具体的信息

7.RabbitMQ与Springboot整合生产端

依赖，参考上面的demo；
编写测试RabbitSender
确认消息，需要一个回调的监听，这个监听就是生产者把消息发出去之后，
Borker收到消息，然后给我们回一个confirm应答,根据应答的状态是成功
还是失败，来确认消息是发送成功还是失败。这样的话需要一个Listener。

代码示例：

public class RabbitSender {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 这里就是确认消息的回调监听，用于确认消息是否被broker所收到
     */
    final RabbitTemplate.ConfirmCallback  confirmCallback = new RabbitTemplate.ConfirmCallback() {
        /**
         *
         * @param correlationData 作为唯一的标识
         * @param ack broker是否落盘成功
         * @param cause 失败的一些异常信息
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        }
    };

    /**
     *  对外发送消息的方法
     * @param message 具体消息的内容
     * @param properties 额外的附加属性
     * @throws Exception
     */
    public void send(Object message, Map<String, Object> properties)throws Exception{

        MessageHeaders mhs = new MessageHeaders(properties);

        //构造消息
         Message<?>  msg = MessageBuilder.createMessage(message,mhs);

         //设置监听
        rabbitTemplate.setConfirmCallback(confirmCallback);

        MessagePostProcessor mpp = new MessagePostProcessor() {

            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message)
                    throws AmqpException {
                System.err.println("---> post to do: " + message);
                return message;
            }
        };
        
        //指定业务唯一的id
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        //发送消息，correlationData作为唯一的标记
        rabbitTemplate.convertAndSend("exchange-1",
                "springboot.rabbit",
                msg,
                mpp,
               correlationData);
    }

8.RabbitMQ与springboot整合-消费端

代码示例：
编写consumer类

@Component
public class RabbitReceive {
   /**
    *     组合使用监听
    *     @RabbitListener @QueueBinding @Queue @Exchange
    * @param message
    * @param channel
    * @throws Exception
    */
   @RabbitListener(bindings = @QueueBinding(
               value = @Queue(value = "queue-1", durable = "true"),
               exchange = @Exchange(name = "exchange-1",
               durable = "true",
               type = "topic",
               ignoreDeclarationExceptions = "true"),
               key = "springboot.*"
            )
         )
   @RabbitHandler
   public void onMessage(Message message, Channel channel) throws Exception {
      // 1. 收到消息以后进行业务端消费处理
      System.err.println("-----------------------");
      System.err.println("消费消息:" + message.getPayload());

      //  2. 处理成功之后 获取deliveryTag 并进行手工的ACK操作, 因为我们配置文件里配置的是 手工签收
      // spring.rabbitmq.listener.simple.acknowledge-mode=manual
      Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
      channel.basicAck(deliveryTag, false);
   }
}

测试类：

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {

   
   @Autowired
   private RabbitSender reRabbitSender;
   
   
   @Test
   public void testSender() throws Exception {
      Map<String, Object> properties = new HashMap<String, Object>();
      properties.put("attr1", "12345");
      properties.put("attr2", "abcde");
      reRabbitSender.send("hello rabbitmq!", properties);
      
      Thread.sleep(10000);
   } 
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1951662.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

The resource type Sheet does not implement java.lang.AutoCloseable

The resource type Sheet does not implement java.lang.AutoCloseable

The resource type Sheet does not implement java.lang.AutoCloseable 修改一下 应该是【高版本JDK】切换集成到我这个项目【低版本 JDK 8】出错了
阅读更多...
Ubuntu设置网络

Ubuntu设置网络

进入网络配置文件夹 cd /etc/netplan 使用 vim 打开下的配置文件 打开后的配置 配置说明&#xff1a; network:# 网络配置部分ethernets:# 配置名为ens33的以太网接口ens33:addresses:# 为ens33接口分配IP地址192.168.220.30&#xff0c;子网掩码为24位- 192.168.220.30/24n…
阅读更多...
【MySQL】：表操作语法大全

【MySQL】：表操作语法大全

表内容的操作 增删改查 CRUD (create、retrieve、update、delete) 新增 基本语法 语法为&#xff1a; insert into 表名 values (值&#xff0c;值&#xff0c;值...);这里的列数和类型&#xff0c;要和表结构匹配插入中文的话&#xff0c;要确保数据库创建的时候要设置字…
阅读更多...
Python 高阶语法

Python 高阶语法

前言&#xff1a; 我们通过上篇文章学习了Python的基础语法&#xff0c;接下来我们来学习Python的高阶语法 1.初识对象 在Python中我们可以做到和生活中那样&#xff0c;设计表格、生产表格、填写表格的组织形式的 面向对象包含 3 大主要特性&#xff1a;  封装  继承 …
阅读更多...
NLP基础知识2【各种大模型的注意力】

NLP基础知识2【各种大模型的注意力】

注意力 传统Attention存在的问题优化方向变体有哪些现在的主要变体集中在KVMulti-Query AttentionGrouped-query AttentionFlashAttention 传统Attention存在的问题 上下文约束速度慢&#xff0c;显存占用大&#xff08;因为注意力考虑整体信息&#xff0c;所以每一个位置都要…
阅读更多...
mysql之触发器的使用

mysql之触发器的使用

cr一&#xff1a;创建goods表和orders表&#xff1b; mysql> use mydb16_tirgeer Database changed mysql> create table goods(-> gid char(8) primary key,-> name varchar(10),-> price decimal(8,2),->-> num int); Query OK, 0 rows affected (0.0…
阅读更多...
18 Python常用内置函数——排序与逆序

18 Python常用内置函数——排序与逆序

sorted() 对列表、元组、字典、集合或其他可迭代对象进行排序并返回新列表&#xff0c;reversed() 对可迭代对象&#xff08;生成器对象和具有惰性求值特性的 zip、map、filter、enumerate 等类似对象除外&#xff09;进行翻转&#xff08;首尾交换&#xff09;并返回可迭代的 …
阅读更多...
《GPT-4o mini：开启开发与创新的新纪元》

《GPT-4o mini：开启开发与创新的新纪元》

在科技发展的快速进程中&#xff0c;OpenAI 推出的 GPT-4o mini 模型如同一阵春风&#xff0c;给开发者们带来了新的希望和机遇。它以其卓越的性能和极具吸引力的价格&#xff0c;成为了行业内热议的焦点。 当我首次听闻 GPT-4o mini 的消息时&#xff0c;内心充满了好奇与期待…
阅读更多...
深度学习目标检测入门实战

深度学习目标检测入门实战

深度学习目标检测入门实战 一、什么是目标检测二、目标检测常用的数据集&#xff08;开源&#xff09;&#xff08;一&#xff09;VOC数据集&#xff08;1&#xff09;背景知识&#xff08;2&#xff09;数据集的下载&#xff08;3&#xff09;VOC2007 数据集的标注&#xff08…
阅读更多...
vue3前端架构---打包配置

vue3前端架构---打包配置

最近看到几篇vue3配置项的文章&#xff0c;转载记录一下 Vue3.2 vue/cli-service 打包 chunk-vendors.js 文件过大导致页面加载缓慢解决方案-CSDN博客文章浏览阅读2k次&#xff0c;点赞8次&#xff0c;收藏9次。Vue3.2 vue/cli-service 打包 chunk-vendors.js 文件过大导致页…
阅读更多...
苦学Opencv的第九天：模板匹配

苦学Opencv的第九天：模板匹配

Python OpenCV入门到精通学习日记&#xff1a;模板匹配 前言 模板匹配是一种最原始、最基本的识别方法&#xff0c;可以在原始图像中寻找特定图像的位置。模板匹配经常应用于简单的图像查找场景中&#xff0c;例如&#xff0c;在集体合照中找到某个人的位置。 #mermaid-svg-N…
阅读更多...
Linux中，MySQL索引、事物与存储引擎

Linux中，MySQL索引、事物与存储引擎

MySQL索引介绍 索引是一个排序的列表,在这个列表中存储着索引的值和包含这个值的数据所在行的物理地址。在数据十分庞大的时候&#xff0c;索引可以大大加快查询的速度。这是因为使用索引后可以不用扫描全表来定位某行的数据,而是先通过索引表找到该行数据对应的物理地址然后访…
阅读更多...
tinyxml2的入门教程

tinyxml2的入门教程

tinyxml2的入门教程 前言一、tinyxml2 创建xml 文件二、tinyxml2 添加数据三、tinyxml2 更改数据四、tinyxml2 删除数据五、tinyxml2 打印总结 前言 xml 是一种标记型文档&#xff0c;有两种基本解析方式&#xff1a;DOM(Document Object Model&#xff0c;文档对象模型)和SAX…
阅读更多...
C++ 数字和数组解析

C++ 数字和数组解析

文章目录 1. 定义数字 2. 数学运算 3. 随机数 4. 数组 声明数组 初始化数组 5. 访问数组元素 6. 数组类型 7. 多维数组 二维数组 初始化二维数组 访问二维数组元素 8. 指向数组的指针 9. 传递数组给函数 10. 从函数返回数组 1. 定义数字 通常&#xff0c;当需要…
阅读更多...
嵌入式人工智能（23-基于树莓派4B的温湿度传感器DHT11）

嵌入式人工智能（23-基于树莓派4B的温湿度传感器DHT11）

1、湿度传感器 目前市面上&#xff0c;仅测量湿度的传感器很少&#xff0c;普遍使用的都是温/湿度传感器&#xff0c;即以温/湿度一体式的探 头作为测温元件&#xff0c;将温度和湿度信号采集出来&#xff0c;再经过稳压滤波、运算放大、非线性校正、V转换、恒流及反向保护等电…
阅读更多...
[Meachines] Lame smbd3.0-RCE

[Meachines] Lame smbd3.0-RCE

信息收集 IP AddressOpening Ports10.10.10.3TCP:21,22,139,445,3632 $ nmap -p- 10.10.10.3 --min-rate 1000 -sC -sV 21/tcp open ftp vsftpd 2.3.4 |_ftp-anon: Anonymous FTP login allowed (FTP code 230) | ftp-syst: | STAT: | FTP server status: | …
阅读更多...
【Stable Diffusion】（基础篇五）—— 使用SD提升分辨率

【Stable Diffusion】（基础篇五）—— 使用SD提升分辨率

使用SD提升分辨率 本系列博客笔记主要参考B站nenly同学的视频教程&#xff0c;传送门&#xff1a;B站第一套系统的AI绘画课&#xff01;零基础学会Stable Diffusion&#xff0c;这绝对是你看过的最容易上手的AI绘画教程 | SD WebUI 保姆级攻略_哔哩哔哩_bilibili 在前期作画的…
阅读更多...
【NoSQL数据库】Redis知识小册

【NoSQL数据库】Redis知识小册

一、缓存穿透 缓存穿透是先查Redis&#xff0c;发现缓存中没有数据&#xff0c;再查数据库。然而&#xff0c;如果查询的数据在数据库中也不存在&#xff0c;那么每次查询都会绕过缓存&#xff0c;直接落到数据库上。 解决方案一、缓存空数据 查询Redis缓存&#xff1a;首先查…
阅读更多...
独立开发者系列（34）——node项目部署

独立开发者系列（34）——node项目部署

本节梳理node项目的部署&#xff0c;搭建一个外部能正常访问的node网站。将开发好的项目搭建到系统里面。Node的部署比PHP版本要复杂一些。部署项目前要理解几个概念。Nodejs版本管理器概念。 NVM概念&#xff0c;我们平时开发是在本地电脑上开发&#xff0c;开发的时候&#x…
阅读更多...
计算机毕业设计：基于SSM的宠物领养系统

计算机毕业设计：基于SSM的宠物领养系统

私信获取完整代码 一、选题背景介绍 &#x1f4d6;☕️&#x1f30a;&#x1f4dd;&#x1f4da;&#x1f3a9;&#x1f680;&#x1f4e3; &#x1f3a9; 宠物领养系统&#xff1a;帮助爱宠人士更好的去查看可以领养的宠物&#xff0c;帮助宣传相关保护宠物相关知识 &…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023