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

本篇内容是本人精心整理；主要讲述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() {@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {System.err.println("------- error ---------");}@Overridepublic 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 {@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 这里就是确认消息的回调监听，用于确认消息是否被broker所收到*/final RabbitTemplate.ConfirmCallback  confirmCallback = new RabbitTemplate.ConfirmCallback() {/**** @param correlationData 作为唯一的标识* @param ack broker是否落盘成功* @param cause 失败的一些异常信息*/@Overridepublic 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() {@Overridepublic org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message)throws AmqpException {System.err.println("---> post to do: " + message);return message;}};//指定业务唯一的idCorrelationData 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.*"))@RabbitHandlerpublic 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=manualLong deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);channel.basicAck(deliveryTag, false);}
}

测试类：

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {@Autowiredprivate RabbitSender reRabbitSender;@Testpublic 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.mzph.cn/news/876727.shtml

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

相关文章

【C++】C++入门知识（上）

【C++】C++入门知识（上）

好久不见&#xff0c;本篇介绍一些C的基础&#xff0c;没有特别的主题&#xff0c;话不多说&#xff0c;直接开始。 1.C的第一个程序 C中需要把定义文件代码后缀改为 .cpp 我们在 test.cpp 中来看下面程序 #include <stdio.h> int main() {printf("hello world\n…
阅读更多...
SQL Server 设置端口号：详细步骤与注意事项

SQL Server 设置端口号：详细步骤与注意事项

目录 一、了解SQL Server端口号的基础知识 1.1 默认端口号 1.2 静态端口与动态端口 二、使用SQL Server配置管理器设置端口号 2.1 打开SQL Server配置管理器 2.2 定位到SQL Server网络配置 2.3 修改TCP/IP属性 2.4 重启SQL Server服务 三、注意事项 3.1 防火墙设置 3…
阅读更多...
Java小抄｜Java中的List与Map转换

Java小抄｜Java中的List与Map转换

文章目录 1 List<User> 转Map<User.id,User>2 基础类型的转换&#xff1a;List < Long> 转 Map<Long,Long> 1 List 转Map<User.id,User> Map<Long, User> userMap userList.stream().collect(Collectors.toMap(User::getId, v -> v, …
阅读更多...
p28 vs环境-C语言实用调试技巧

p28 vs环境-C语言实用调试技巧

int main() { int i0; for(i0;i<100;i) { printf("%d",i); } } 1.Debug 和Release的介绍 Debug通常称为调试版本&#xff0c;它包含调试信息&#xff0c;并且不做任何优化&#xff0c;便于程序员调试程序。 Release称为发布版本&#x…
阅读更多...
PTPD 在 QNX 系统上的授时精度验证与误差排查

PTPD 在 QNX 系统上的授时精度验证与误差排查

文章目录 0. 引言1.关键函数实现2. 验证策略与结果3. 授时误差的排查与解决3. 授时误差的排查与解决4. 结论 0. 引言 PTPD是一种时间同步的开源实现&#xff0c;在不同操作系统上的表现可能存在显著差异。 本文通过在QNX系统上运行PTPD&#xff0c;针对其授时精度进行详细验证…
阅读更多...
探索算法系列 - 双指针

探索算法系列 - 双指针

目录 移动零&#xff08;原题链接&#xff09; 复写零&#xff08;原题链接&#xff09; 快乐数&#xff08;原题链接&#xff09; 盛最多水的容器&#xff08;原题链接&#xff09; 有效三角形的个数&#xff08;原题链接&#xff09; 查找总价格为目标值的两个商品&…
阅读更多...
优化算法：2.粒子群算法(PSO)及Python实现

优化算法：2.粒子群算法(PSO)及Python实现

一、定义 粒子群算法&#xff08;Particle Swarm Optimization&#xff0c;PSO&#xff09;是一种模拟鸟群觅食行为的优化算法。想象一群鸟在寻找食物&#xff0c;每只鸟都在尝试找到食物最多的位置。它们通过互相交流信息&#xff0c;逐渐向食物最多的地方聚集。PSO就是基于这…
阅读更多...
【python_将一个列表中的几个字典改成二维列表，并删除不需要的列】

【python_将一个列表中的几个字典改成二维列表，并删除不需要的列】

def 将一个列表中的几个字典改成二维列表(original_list,headersToRemove_list):# 初始化一个列表用于存储遇到的键&#xff0c;保持顺序ordered_keys []# 遍历data中的每个字典&#xff0c;添加其键到ordered_keys&#xff0c;如果该键还未被添加for d in original_list:for …
阅读更多...
P4009 汽车加油行驶问题题解

P4009 汽车加油行驶问题题解

P4009 汽车加油行驶问题 紫题&#xff0c;但是DFS。 思路 记忆化搜索&#xff0c;分多钟情况去搜索。 注意该题不用标记&#xff0c;有可能会往回走。 有可能这样走。 代码 #include<bits/stdc.h> #include<cstring> #include<queue> #include<set&g…
阅读更多...
redis:清除缓存的最简单命令示例

redis:清除缓存的最简单命令示例

清除redis缓存命令(执行命令列表见截图) 1.打开cmd窗口&#xff0c;并cd进入redis所在目录 2.登录redis redis-cli 3.查询指定队列当前的记录数 llen 队列名称 4.清除指定队列所有记录 ltrim 队列名称 1 0 5.再次查询&#xff0c;确认队列的记录数是否已清除
阅读更多...
配置和连接另一台电脑上的 MySQL 数据库

配置和连接另一台电脑上的 MySQL 数据库

要配置和连接另一台电脑上的 MySQL 数据库&#xff0c;可以按照以下步骤进行设置&#xff1a; 1. 配置 MySQL 服务器 在目标计算机上&#xff08;192.168.10.103&#xff09;进行以下操作&#xff1a; 修改 MySQL 配置文件&#xff1a; 打开 MySQL 配置文件&#xff08;通常位…
阅读更多...
【系统架构设计师】十八、信息系统架构设计理论与实践①

【系统架构设计师】十八、信息系统架构设计理论与实践①

目录 一、信息系统架构概述 二、信息系统架构风格与分类 2.1 信息系统架构风格 2.2 信息系统架构分类 三、信息系统架构模型 3.1 单体应用 3.2 客户机/服务器 3.2.1 二层 C/S 3.2.2 三层 C/S 和 B/S 3.2.3 多层 C/S 和 B/S 3.2.4 MVC 3.3 面向服务架构(SOA)模式 …
阅读更多...
Activiti 本地画流程 http://localhost:8080/activiti-app/#/

Activiti 本地画流程 http://localhost:8080/activiti-app/#/

http://localhost:8080/activiti-app/#/ 1、本地安装了Tomcat 2、本地安装了Activiti 3、拷贝Activiti中这两个文件到Tomcat中的webapps目录下 4、启动startu.bat 5、http://localhost:8080/activiti-app/#/ 账号&#xff1a;admin 密码&#xff1a;test
阅读更多...
乐鑫 Matter 技术体验日回顾｜全面 Matter 解决方案驱动智能家居新未来

乐鑫 Matter 技术体验日回顾｜全面 Matter 解决方案驱动智能家居新未来

日前&#xff0c;乐鑫信息科技 (688018.SH) 在深圳成功举办了 Matter 方案技术体验日活动&#xff0c;吸引了众多照明电工、窗帘电机、智能门锁、温控等智能家居领域的客户与合作伙伴。活动现场&#xff0c;乐鑫产研团队的小伙伴们与来宾围绕 Matter 产品研发、测试认证、生产工…
阅读更多...
Python学习笔记46：游戏篇之外星人入侵(七)

Python学习笔记46：游戏篇之外星人入侵(七)

前言 到目前为止&#xff0c;我们已经完成了游戏窗口的创建&#xff0c;飞船的加载&#xff0c;飞船的移动&#xff0c;发射子弹等功能。很高兴的说一声&#xff0c;基础的游戏功能已经完成一半了&#xff0c;再过几天我们就可以尝试驾驶 飞船击毁外星人了。当然&#xff0c;计…
阅读更多...
解析西门子PLC的String和WString

解析西门子PLC的String和WString

西门子PLC有两种字符串类型&#xff0c;String与WString String 用于存放英文数字标点符号等ASCII字符&#xff0c;每个字符占用一个字节 WString宽字符串用于存放中文、英文、数字等Unicode字符&#xff0c;每个字符占用两个字节 之前我搞过一篇解析String的 关于使用TCP-…
阅读更多...
Vue3 Pinia的创建与使用代替Vuex 全局数据共享 同步异步

Vue3 Pinia的创建与使用代替Vuex 全局数据共享 同步异步

介绍 提供跨组件和页面的共享状态能力&#xff0c;作为Vuex的替代品&#xff0c;专为Vue3设计的状态管理库。 Vuex&#xff1a;在Vuex中&#xff0c;更改状态必须通过Mutation或Action完成&#xff0c;手动触发更新。Pinia&#xff1a;Pinia的状态是响应式的&#xff0c;当状…
阅读更多...
Linux内核 mmap内存映射的实现原理

Linux内核 mmap内存映射的实现原理

在Linux内核以及Linux系统编程的时候&#xff0c;经常会碰到mmap内存映射&#xff0c;mmap函数是实现高性能编程的一个关键点。本文详细介绍一下mmap实现原理。 虚拟地址映射物理地址 虚拟地址映射物理地址采用的是页表机制&#xff0c;64位CPU采用的是4级页表。 64位CPU虚拟…
阅读更多...
鸿蒙 HarmonyOS NEXT端云一体化开发-认证服务篇

鸿蒙 HarmonyOS NEXT端云一体化开发-认证服务篇

一、开通认证服务 地址&#xff1a;AppGallery Connect (huawei.com) 步骤&#xff1a; 1 进入到项目设置页面中&#xff0c;并点击左侧菜单中的认证服务 2 选择需要开通的服务并开通二、端侧项目环境配置 添加依赖 entry目录下的oh-package.json5 // 添加&#xff1a;主要前…
阅读更多...
《python程序语言设计》第6章14题 估算派值 类似莱布尼茨函数。但是我看不明白

《python程序语言设计》第6章14题 估算派值 类似莱布尼茨函数。但是我看不明白

这个题提供的公式我没看明白&#xff0c;后来在网上找到了莱布尼茨函数 c 0 for i in range(1, 902, 100):a (-1) ** (i 1)b 2 * i - 1c a / bprint(i, round(4 / c, 3))结果 #按题里的信息&#xff0c;但是结果不对&#xff0c;莱布尼茨函数到底怎么算呀。
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023