0


快速搞定分布式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  =newConnectionFactory();
connectionFactory.setHost("192.168.56.107");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");//2.创建ConnectionFactoryConnection connection = connectionFactory.newConnection();//3.创建channelChannel channel = connection.createChannel();// 4 声明String queueName ="testone";///  参数: queue名字,是否持久化,独占的queue(仅供此连接),不使用时
是否自动删除, 其他参数
channel.queueDeclare(queueName,false,false,false,null);Map<String,Object> headers =newHashMap<>();//deliveryMode 1:消息非持久化,2:消息持久化AMQP.BasicProperties props =newAMQP.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 =newConnectionFactory();

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 =newQueueingConsumer(channel);//参数:队列名称,是否自动ACK,Consumer,相当于设置监听
channel.basicConsume(queueName,true,consumer);//循环获取消息while(true){//获取消息,如果没有消息,这一步将会一直阻塞Delivery delivery = consumer.nextDelivery();String msg =newString(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(newConfirmListener(){@OverridepublicvoidhandleNack(long deliveryTag,boolean multiple)throwsIOException{System.err.println("------- error ---------");}@OverridepublicvoidhandleAck(long deliveryTag,boolean multiple)throwsIOException{System.err.println("------- ok ---------");}});
      
channel.basicPublish(exchangeName, routingKey1 ,null, msg.getBytes());

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

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

//5 监听
channel.addReturnListener(newReturnListener(){publicvoidhandleReturn(int replyCode,String replyText,String exchange,String routingKey,AMQP.BasicProperties properties,byte[] body)throwsIOException{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: "+newString(body));}});

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

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

VoidBasicQos(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 =newString(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 创建ConnectionFactoryConnectionFactory connectionFactory =newConnectionFactory();
connectionFactory.setHost("192.168.56.107");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();//4 声明String exchangeName ="test_dlx_exchange";String routingKey ="group.bfxy";//5 发送Map<String,Object> headers =newHashMap<String,Object>();AMQP.BasicProperties props =newAMQP.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。

代码示例:

publicclassRabbitSender{@AutowiredprivateRabbitTemplate rabbitTemplate;/**
     * 这里就是确认消息的回调监听,用于确认消息是否被broker所收到
     */finalRabbitTemplate.ConfirmCallback  confirmCallback =newRabbitTemplate.ConfirmCallback(){/**
         *
         * @param correlationData 作为唯一的标识
         * @param ack broker是否落盘成功
         * @param cause 失败的一些异常信息
         */@Overridepublicvoidconfirm(CorrelationData correlationData,boolean ack,String cause){}};/**
     *  对外发送消息的方法
     * @param message 具体消息的内容
     * @param properties 额外的附加属性
     * @throws Exception
     */publicvoidsend(Object message,Map<String,Object> properties)throwsException{MessageHeaders mhs =newMessageHeaders(properties);//构造消息Message<?>  msg =MessageBuilder.createMessage(message,mhs);//设置监听
        rabbitTemplate.setConfirmCallback(confirmCallback);MessagePostProcessor mpp =newMessagePostProcessor(){@Overridepublicorg.springframework.amqp.core.MessagepostProcessMessage(org.springframework.amqp.core.Message message)throwsAmqpException{System.err.println("---> post to do: "+ message);return message;}};//指定业务唯一的idCorrelationData correlationData =newCorrelationData(UUID.randomUUID().toString());//发送消息,correlationData作为唯一的标记
        rabbitTemplate.convertAndSend("exchange-1","springboot.rabbit",
                msg,
                mpp,
               correlationData);}

8.RabbitMQ与springboot整合-消费端

代码示例:
编写consumer类

@ComponentpublicclassRabbitReceive{/**
    *     组合使用监听
    *     @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.*"))@RabbitHandlerpublicvoidonMessage(Message message,Channel channel)throwsException{// 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)@SpringBootTestpublicclassApplicationTests{@AutowiredprivateRabbitSender reRabbitSender;@TestpublicvoidtestSender()throwsException{Map<String,Object> properties =newHashMap<String,Object>();
      properties.put("attr1","12345");
      properties.put("attr2","abcde");
      reRabbitSender.send("hello rabbitmq!", properties);Thread.sleep(10000);}}
标签: 分布式 rabbitmq

本文转载自: https://blog.csdn.net/TOMORROW6COME/article/details/140710509
版权归原作者 TOMORROW6COME 所有, 如有侵权,请联系我们删除。

“快速搞定分布式RabbitMQ---RabbitMQ进阶与实战”的评论:

还没有评论