Centos7安装RabbitMQ
第一步、安装任何软件包之前，建议使用以下命令更新软件包和存储库
yum -y update
第二步、Erlang在默认的YUM存储库中不可用，因此您将需要安装EPEL存储库
yum -yinstall epel-release
yum -y update
也可以去Erlang官网下载Erlang存储库https://packagecloud.io/rabbitmq/erlang?page=1，将下载好的rpm包防至Centos上
第三步、RabbitMQ是基于Erlang（面向高并发的语言）语言开发，所以在安装RabbitMQ之前，需要先安装Erlang语言
yum -yinstall erlang socat
第四步、检查Erlang版本，默认是最新版本
erl -version
第五步、下载RabbitMQ
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/rabbitmq_v3_6_16/rabbitmq-server-3.6.16-1.el7.noarch.rpm
也可以直接去GitHub上下载https://www.rabbitmq.com/download.html，将下载好的rpm包放至Centos上
第六步、安装RabbitMQ
rpm-ivh rabbitmq-server-3.6.16-1.el7.noarch.rpm
# 或者 rpm -Uvh rabbitmq-server-3.6.16-1.el7.noarch.rpm# -U表升级
第七步、安装管理界面的插件(没有这个插件无法启动管理界面)
# 开启管理界面
rabbitmq-plugins enable rabbitmq_management
第八步、开放5672和15672端口
firewall-cmd --zone=public --add-port=5672/tcp --permanent&& firewall-cmd --reload
firewall-cmd --zone=public --add-port=15672/tcp --permanent&& firewall-cmd --reload
恭喜你，RabbitMQ现已经安装在系统上！
第九步、启动RabbitMQ服务
systemctl start rabbitmq-server #启动服务
systemctl status rabbitmq-server #查看服务状态
systemctl stop rabbitmq-server #停止服务
systemctl enable rabbitmq-server #开启启动服务
第十步、默认的账号密码仅限于本机 localhost 进行访问，所以需要添加一个远程登录的用户
# 创建账号和密码
rabbitmqctl add_user 用户名 密码
# 设置用户角色：administrator
rabbitmqctl set_user_tags 用户名 角色
# 为用户添加资源权限，添加配置、写、读权限
rabbitmqctl set_permissions -p"/" 用户名 ".*"".*"".*"#修改用户密码
rabbitmqctl change_password 用户名 新密码
RabbitMQ的使用
RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控制、负载均衡等特性，使得RabbitMQ拥有更加广泛的应用场景。RabbitMQ跟Erlang和AMQP有关。下面简单介绍一下Erlang和AMQP。
Erlang：
Erlang是一门动态类型的函数式编程语言，它也是一门解释型语言，由Erlang虚拟机解释执行。从语言模型上说，Erlang是基于Actor模型的实现。在Actor模型里面，万物皆Actor，每个Actor都封装着内部状态，Actor相互之间只能通过消息传递这一种方式来进行通信。对应到Erlang里，每个Actor对应着一个Erlang进程，进程之间通过消息传递进行通信。相比共享内存，进程间通过消息传递来通信带来的直接好处就是消除了直接的锁开销(不考虑Erlang虚拟机底层实现中的锁应用)。
AMQP：
AMQP(Advanced Message Queue Protocol)定义了一种消息系统规范。这个规范描述了在一个分布式的系统中各个子系统如何通过消息交互。而RabbitMQ则是AMQP的一种基于erlang的实现。AMQP将分布式系统中各个子系统隔离开来，子系统之间不再有依赖。子系统仅依赖于消息。子系统不关心消息的发送者，也不关心消息的接受者
一、RabbitMQ的作用：
1、削峰填谷
2、解耦
3、平衡上下游的处理速度
二. rabbitmq基本原理
RabbitMQ是消息队列的一种实现，那么一个消息队列到底需要什么？答案是队列，即Queue，那么接下来所有名词都是围绕这个Queue来拓展的。
ConnectionFactory：
Connection：
Chanel：
因为TCP连接的建立和释放都是十分昂贵的，
Exchange：
分发消息
Queue：
BindingKey：
RoutingKey：
RabbitMQ的交换机类型：
Fanout类型交换机——扇出交换机
Direct类型交换机——直连交换机
Topic类型交换机——主题交换机
RabbitMQ的消息模型：
简单模型-Simple ============================================================
简单模型也叫直连模型，一个生产者一个消费者
获取连接的工具类：
packagecom.example.demo.rabbitMQ;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassRabbitMQUtils{privatestaticfinalConnectionFactory connectionFactory =newConnectionFactory();static{
connectionFactory.setHost("192.168.119.134");
connectionFactory.setPort(5672);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");}publicstaticConnectiongetConnection(){try{return connectionFactory.newConnection();}catch(IOException e){thrownewRuntimeException(e);}catch(TimeoutException e){thrownewRuntimeException(e);}}}
生产者：
packagecom.example.demo.rabbitMQ.simple;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;publicclassProducer{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名 == BindingKey
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/
channel.queueDeclare("simple-queue",true,false,false,null);/**
* 第一个参数:交换机名称，如果为“”空串表示使用默认交换机，将消息投递到默认交换机，通过默认交换机投递到与之绑定的队列
* 第二个参数:RoutingKey==BindingKey，在简单模型中绑定在默认的交换机上的队列，BindingKey和队列名一致
* 第三个参数:消息的属性(消息是否持久化、消息存活时间)
* 第四个参数:消息内容
*/
channel.basicPublish("","simple-queue",null,"Hellow World ".getBytes());}}
消费者：
packagecom.example.demo.rabbitMQ.simple;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumer{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 为了保险起见，防止生产方未启动队列未创建的情况下消费方启动后报404异常，最好在消费方中也声明创建队列，注意消费方和生产方声明的队列必须完全一致
* 注意：不用考虑队列是否会重复创建，在RabbitMQ中如果队列已经存在是不会被重新创建的
*/
channel.queueDeclare("simple-queue",true,false,false,null);/**
* 第一个参数:队列名,所要消费的队列
* 第二个参数:是否自动确认
* true表示自送确认:消息拿到了就确认
* false表示手动确认:消息处理完成后确认
*/
channel.basicConsume("simple-queue",true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body)+envelope.getDeliveryTag());}});}}
工作模型-Work ============================================================
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度，消息在队列中就会堆积越来越多，无法及时处理。为了平衡上下游的处理速度（生产速度大于消费速度）就可以使用工作模型：让多个消费者绑定到一个队列，共同消费队列中的消息提升消费速度。
队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的
。
生产者：
packagecom.example.demo.rabbitMQ.work;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;publicclassProducer{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名 == BindingKey
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/
channel.queueDeclare("work-queue",true,false,false,null);for(int i =0;i <=1000;i++){/**
* 第一个参数:交换机名称，如果为“”空串表示使用默认交换机，将消息投递到默认交换机，通过默认交换机投递到与之绑定的队列
* 第二个参数:RoutingKey==BindingKey，在工作模型中绑定在默认的交换机上的队列，BindingKey和队列名一致
* 第三个参数:消息的属性(消息是否持久化、消息存活时间)
* 第四个参数:消息内容
*/
channel.basicPublish("","work-queue",null,("Hello World"+ i +" ").getBytes());}}}
消费者1：
packagecom.example.demo.rabbitMQ.work;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumerOne{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 为了保险起见，防止生产方未启动队列未创建的情况下消费方启动后报404异常，最好在消费方中也声明创建队列，注意消费方和生产方声明的队列必须完全一致
* 注意：不用考虑队列是否会重复创建，在RabbitMQ中如果队列已经存在是不会被重新创建的
*/
channel.queueDeclare("work-queue",true,false,false,null);/**
* 设置每次抓取的数据条数
* 不设置：默认平分队列中的消息，如果队列中中有100条数据，那么两个消费方各抓取50条数据进行消费
* 设置 ：按照设置的条数抓取，如果设置1，那么消费方从队列中每次抓取1条数据进行消费，消费完成后再抓取1条，直到队列中没有消息
*/
channel.basicQos(1);/**
* 第一个参数:队列名,所要消费的队列
* 第二个参数:是否自动确认
* true表示自送确认:消息拿到了就确认
* false表示手动确认:消息处理完成后确认
*/
channel.basicConsume("work-queue",false,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body)+envelope.getDeliveryTag());/**
* 手动确认方式
* 第一个参数:包裹的标签（消息的标签，RabbitMQ将每个消息看成是一个包裹），是个整数
* 第二个参数:是否多条消息批量确认，如果第一、二、三...条消息没有确认，后面一条消息确认被消费了，那么前面所有的消息都会被确认消费了
*/
channel.basicAck(envelope.getDeliveryTag(),false);}});}}
消费者2：
packagecom.example.demo.rabbitMQ.work;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumerTwo{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 为了保险起见，防止生产方未启动队列未创建的情况下消费方启动后报404异常，最好在消费方中也声明创建队列，注意消费方和生产方声明的队列必须完全一致
* 注意：不用考虑队列是否会重复创建，在RabbitMQ中如果队列已经存在是不会被重新创建的
*/
channel.queueDeclare("work-queue",true,false,false,null);/**
* 设置每次抓取的数据条数
* 不设置：默认平分队列中的消息，如果队列中中有100条数据，那么两个消费方各抓取50条数据进行消费
* 设置 ：按照设置的条数抓取，如果设置1，那么消费方从队列中每次抓取1条数据进行消费，消费完成后再抓取1条，直到队列中没有消息
*/
channel.basicQos(1);/**
* 第一个参数:队列名,所要消费的队列
* 第二个参数:是否自动确认
* true表示自送确认:消息拿到了就确认
* false表示手动确认:消息处理完成后确认
*/
channel.basicConsume("work-queue",false,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{try{Thread.sleep(500);}catch(InterruptedException e){thrownewRuntimeException(e);}System.out.println(newString(body)+envelope.getDeliveryTag());/**
* 手动确认方式
* 第一个参数:包裹的标签（消息的标签，RabbitMQ将每个消息看成是一个包裹），是个整数
* 第二个参数:是否多条消息批量确认，如果第一、二、三...条消息没有确认，后面一条消息确认被消费了，那么前面所有的消息都会被确认消费了
*/
channel.basicAck(envelope.getDeliveryTag(),false);}});}}
广播模型-Fanout ============================================================
在广播模式下，生产者将消息发送给交换机后，交换机将消息推送给每一个绑定在交换机上的队列，每一个队列的消费者都能拿到消息。在Fanout模式中，一条消息，会被所有绑定在该交换机上的队列消费。
生产者：
packagecom.example.demo.rabbitMQ.fanout;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;publicclassProducer{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 声明一个Fanout类型的交换机
* 第一个参数:交换机名称
* 第二个参数:交换机类型:BuiltinExchangeType.FANOUT、BuiltinExchangeType.TOPIC、BuiltinExchangeType.DIRECT
*/
channel.exchangeDeclare("fanout-exchange",BuiltinExchangeType.FANOUT);/**
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/
channel.queueDeclare("fanout-queue",true,false,false,null);/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey
*/
channel.queueBind("fanout-queue","fanout-exchange","fanout-model");/**
* 第一个参数:交换机名称，如果为“”空串表示使用默认交换机，将消息投递到默认交换机，通过默认交换机投递到与之绑定的队列
* 第二个参数:RoutingKey==BindingKey，在Fanout模式下指定RoutingKey没有意义，Fanout模式下每个与该交换机绑定的队列都能拿到消息
* 第三个参数:消息的属性(消息是否持久化、消息存活时间)
* 第四个参数:消息内容
*/
channel.basicPublish("fanout-exchange","fanout-model",null,"Hellow World ".getBytes());}}
消费者1：
packagecom.example.demo.rabbitMQ.fanout;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumerOne{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 声明一个Fanout类型的交换机
* 第一个参数:交换机名称
* 第二个参数:交换机类型
*/
channel.exchangeDeclare("fanout-exchange",BuiltinExchangeType.FANOUT);/**
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，连接关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
* getQueue()方法可获取队列名称
*/String queueName = channel.queueDeclare("fanout-queue-one",true,false,false,null).getQueue();/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey
*/
channel.queueBind(queueName,"fanout-exchange","fanout-model-one");/**
* 从队列中消费消息
* 第一个参数:队列名称
* 第二个参数:是否自动确认
* 第三个参数:消费者
*/
channel.basicConsume("fanout-queue-one",false,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body)+"One"+" "+envelope.getDeliveryTag());/**
* 确认消息是否消费，给队列反馈
* 第一个参数:包裹（消息）标签
* 第二个参数:是否多条消息批量确认，如果第一、二、三...条消息没有确认，后面一条消息确认被消费了，那么前面所有的消息都会被确认消费了
*/
channel.basicAck(envelope.getDeliveryTag(),true);}});}}
消费者2：
packagecom.example.demo.rabbitMQ.fanout;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumerTwo{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 声明一个Fanout类型的交换机
* 第一个参数:交换机名称
* 第二个参数:交换机类型
*/
channel.exchangeDeclare("fanout-exchange",BuiltinExchangeType.FANOUT);/**
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/
channel.queueDeclare("fanout-queue-two",true,false,false,null);/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey
*/
channel.queueBind("fanout-queue-two","fanout-exchange","fanout-model-two");/**
* 从队列中消费消息
* 第一个参数:队列名称
* 第二个参数:是否自动确认
* 第三个参数:消费者
*/
channel.basicConsume("fanout-queue-two",true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body)+"Two");}});}}
消费者3：
packagecom.example.demo.rabbitMQ.fanout;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumerThree{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 声明一个Fanout类型的交换机
* 第一个参数:交换机名称
* 第二个参数:交换机类型
*/
channel.exchangeDeclare("fanout-exchange",BuiltinExchangeType.FANOUT);/**
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，连接关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
* getQueue()方法可获取队列名称
*/String queueName = channel.queueDeclare("fanout-queue-one",true,false,false,null).getQueue();/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey
*/
channel.queueBind(queueName,"fanout-exchange","fanout-model-one");/**
* 设置每次抓取的数据条数
* 不设置：默认平分队列中的消息，如果队列中中有100条数据，那么两个消费方各抓取50条数据进行消费
* 设置 ：按照设置的条数抓取，如果设置1，那么消费方从队列中每次抓取1条数据进行消费，消费完成后再抓取1条，直到队列中没有消息
*/
channel.basicQos(1);/**
* 从队列中消费消息
* 第一个参数:队列名称
* 第二个参数:是否自动确认
* 第三个参数:消费者
*/
channel.basicConsume(queueName,true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body)+"Three"+" "+envelope.getDeliveryTag());}});}}
定向模型-Direct ============================================================
在定向模型中，生产者将消息发送给交换机后，交换机将消息根据RoutingKey\BindingKey推送到对应的队列上。Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key
进行判断，只有队列的
BindingKey
与消息的
RoutingKey
完全一致，才会接收到消息
生产者：
packagecom.example.demo.rabbitMQ.direct;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importjava.nio.charset.StandardCharsets;publicclassProducer{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();
channel.exchangeDeclare("direct-exchange",BuiltinExchangeType.DIRECT);//队列1String queue1Name = channel.queueDeclare("direct-queue-one",true,false,false,null).getQueue();
channel.queueBind(queue1Name,"direct-exchange","direct-model-one");//队列2String queue2Name = channel.queueDeclare("direct-queue-two",true,false,false,null).getQueue();
channel.queueBind(queue2Name,"direct-exchange","direct-model-two");
channel.basicPublish("direct-exchange","direct-model-two",null,"JAVA是世界上最好的语言 ".getBytes(StandardCharsets.UTF_8));}}
消费者1：
packagecom.example.demo.rabbitMQ.direct;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.nio.charset.StandardCharsets;publicclassConsumerOne{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();
channel.exchangeDeclare("direct-exchange",BuiltinExchangeType.DIRECT);String queueName = channel.queueDeclare("direct-queue-one",true,false,false,null).getQueue();
channel.queueBind(queueName,"direct-exchange","direct-model-one");
channel.basicConsume(queueName,true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body,StandardCharsets.UTF_8)+"One");}});}}
消费者2：
packagecom.example.demo.rabbitMQ.direct;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassConsumerTwo{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();
channel.exchangeDeclare("direct-exchange",BuiltinExchangeType.DIRECT);String queueName = channel.queueDeclare("direct-queue-two",true,false,false,null).getQueue();
channel.queueBind(queueName,"direct-exchange","direct-model-two");
channel.basicConsume(queueName,true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body)+"Two");}});}}
主题模型-Topic ============================================================
在主题模式下，生产者将消息发送给交换机后，交换机根据RoutingKey\BindingKey匹配队列，将消息推送到匹配的队列上。在topic模式下通过
#
号和
*
号进行模糊匹配,通过.进行分割,
#
号表示可能有一个或多个单词，也可能没有；
*
号表示有且仅有一个单词
生产者：
packagecom.example.demo.rabbitMQ.topic;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;publicclassProducer{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 声明一个TOPIC类型的交换机
* 第一个参数:交换机名称
* 第二个参数:交换机类型:BuiltinExchangeType.FANOUT、BuiltinExchangeType.TOPIC、BuiltinExchangeType.DIRECT
*/
channel.exchangeDeclare("topic-exchange",BuiltinExchangeType.TOPIC);/**
* 声明队列1:
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/String queue1Name = channel.queueDeclare("topic-queue-one",true,false,false,null).getQueue();/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey,在topic模式下通过#和*进行模糊匹配,通过.进行分割,#号表示可能有一个或多个单词，也可能没有；*号表示有且仅有一个单词
*/
channel.queueBind(queue1Name,"topic-exchange","topic-model.#");/**
* 声明队列2:
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/String queue2Name = channel.queueDeclare("topic-queue-two",true,false,false,null).getQueue();/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey,在topic模式下通过#和*进行模糊匹配,通过.进行分割,#号表示可能有一个或多个单词，也可能没有；*号表示有且仅有一个单词
*/
channel.queueBind(queue2Name,"topic-exchange","topic-model.*");/**
* 声明队列3:
* 注意：通过channel.queueDeclare()方法创建的Queue绑定在默认的交换机上，且BindingKey和队列名一致
* 第一个参数:队列名
* 第二个参数:队列是否持久化，如果为false，rabbitMQ服务关闭队列消失
* 第三个参数:队列是否为排他队列，如果为true，队列仅供创建它的连接使用，当前连接关闭队列消失
* 第四个参数:队列是否自动删除，如果为true，队列中的消息消费完成，并且消费方关闭后，队列自动删除
*/String queue3Name = channel.queueDeclare("topic-queue-three",true,false,false,null).getQueue();/**
* 将队列绑定到交换机上
* 第一个参数:队列名称
* 第二个参数:交换机名称
* 第三个参数:BindingKey,在topic模式下通过#和*进行模糊匹配,通过.进行分割,#号表示可能有一个或多个单词，也可能没有；*号表示有且仅有一个单词
*/
channel.queueBind(queue3Name,"topic-exchange","topic-model.three.#");/**
* 第一个参数:交换机名称，如果为“”空串表示使用默认交换机，将消息投递到默认交换机，通过默认交换机投递到与之绑定的队列
* 第二个参数:RoutingKey==BindingKey，在topic模式下通过#和*进行模糊匹配,通过.进行分割,#号表示可能有一个或多个单词，也可能没有；*号表示有且仅有一个单词
* 第三个参数:消息的属性(消息是否持久化、消息存活时间)
* 第四个参数:消息内容
*/
channel.basicPublish("topic-exchange","topic-model.three.java",null,"java是世界上最优秀的语言".getBytes());}}
消费者1：
packagecom.example.demo.rabbitMQ.topic;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.nio.charset.StandardCharsets;publicclassConsumerOne{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 为了保险起见，防止生产方未启动队列\交换机未创建消的情况下费方启动后报404异常，最好在消费方中也声明创建队列，注意消费方和生产方声明的队列\交换机必须完全一致
* 注意：不用考虑队列\交换机是否会重复创建，在RabbitMQ中如果队列已经存在是不会被重新创建的
*///声明交换机
channel.exchangeDeclare("topic-exchange",BuiltinExchangeType.TOPIC);//声明队列String queue1Name = channel.queueDeclare("topic-queue-one",true,false,false,null).getQueue();//将队列绑定到交换机上
channel.queueBind(queue1Name,"topic-exchange","topic-model.#");//设置每次抓取的数据条数
channel.basicQos(1);/**
* 第一个参数:被消费的队列名
* 第二个参数:是否自动确认
* 第三个参数:使用默认的消费者
*/
channel.basicConsume(queue1Name,true,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body,StandardCharsets.UTF_8));//在手动确认的情况下使用/**
* 确认消息是否消费，给队列反馈
* 第一个参数:包裹（消息）标签
* 第二个参数:是否多条消息批量确认，如果第一、二、三...条消息没有确认，后面一条消息确认被消费了，那么前面所有的消息都会被确认消费了
*///channel.basicAck(envelope.getDeliveryTag(),false);}});}}
消费者2：
packagecom.example.demo.rabbitMQ.topic;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.nio.charset.StandardCharsets;publicclassConsumerTwo{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 为了保险起见，防止生产方未启动队列\交换机未创建消的情况下费方启动后报404异常，最好在消费方中也声明创建队列，注意消费方和生产方声明的队列\交换机必须完全一致
* 注意：不用考虑队列\交换机是否会重复创建，在RabbitMQ中如果队列已经存在是不会被重新创建的
*///声明交换机
channel.exchangeDeclare("topic-exchange",BuiltinExchangeType.TOPIC);//声明队列String queue2Name = channel.queueDeclare("topic-queue-two",true,false,false,null).getQueue();//将队列绑定到交换机上
channel.queueBind(queue2Name,"topic-exchange","topic-model.*");//设置每次抓取的数据条数
channel.basicQos(1);/**
* 第一个参数:被消费的队列名
* 第二个参数:是否自动确认
* 第三个参数:使用默认的消费者
*/
channel.basicConsume(queue2Name,false,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body,StandardCharsets.UTF_8));//在手动确认的情况下使用/**
* 确认消息是否消费，给队列反馈
* 第一个参数:包裹（消息）标签
* 第二个参数:是否多条消息批量确认，如果第一、二、三...条消息没有确认，后面一条消息确认被消费了，那么前面所有的消息都会被确认消费了
*/
channel.basicAck(envelope.getDeliveryTag(),false);}});}}
消费者3：
packagecom.example.demo.rabbitMQ.topic;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.nio.charset.StandardCharsets;publicclassConsumerThree{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/**
* 为了保险起见，防止生产方未启动队列\交换机未创建消的情况下费方启动后报404异常，最好在消费方中也声明创建队列，注意消费方和生产方声明的队列\交换机必须完全一致
* 注意：不用考虑队列\交换机是否会重复创建，在RabbitMQ中如果队列已经存在是不会被重新创建的
*///声明交换机
channel.exchangeDeclare("topic-exchange",BuiltinExchangeType.TOPIC);//声明队列String queue3Name = channel.queueDeclare("topic-queue-two",true,false,false,null).getQueue();//将队列绑定到交换机上
channel.queueBind(queue3Name,"topic-exchange","topic-model");//设置每次抓取的数据条数
channel.basicQos(1);/**
* 第一个参数:被消费的队列名
* 第二个参数:是否自动确认
* 第三个参数:使用默认的消费者
*/
channel.basicConsume(queue3Name,false,newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println(newString(body,StandardCharsets.UTF_8));//在手动确认的情况下使用/**
* 确认消息是否消费，给队列反馈
* 第一个参数:包裹（消息）标签
* 第二个参数:是否多条消息批量确认，如果第一、二、三...条消息没有确认，后面一条消息确认被消费了，那么前面所有的消息都会被确认消费了
*/
channel.basicAck(envelope.getDeliveryTag(),false);}});}}
SpringBoot+RabbitMQ
第一步：先导依赖
<!--============================================RabbitMQ相关依赖开始============================================--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.4</version></dependency><!--============================================RabbitMQ相关依赖结束============================================-->
第二步：配置RabbitMQ
spring:rabbitmq:host: 192.168.119.134
port:5672username: rabbit
password:123456
第三步：配置RabbitMQ使用的序列化
// 消息的消费方json数据的反序列化@BeanpublicRabbitListenerContainerFactory<?>rabbitListenerContainerFactory(ConnectionFactory connectionFactory){SimpleRabbitListenerContainerFactory factory =newSimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(newJackson2JsonMessageConverter());return factory;}// 定义使用json的方式转换数据@BeanpublicRabbitTemplaterabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate amqpTemplate =newRabbitTemplate();
amqpTemplate.setConnectionFactory(connectionFactory);
amqpTemplate.setMessageConverter(newJackson2JsonMessageConverter());return amqpTemplate;}
第四步：根据情况选择合适的模型
一、简单模型 ============================================================
特点：队列是绑定在默认交换机上，BindingKey就是队列名
消息投递方：
在SpringBoot中即可使用RabbitTemplate进行消息的投递，也可使用原生的方式进行消息投递
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.simple;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/rabbit")publicclassProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;@GetMapping("simple")publicStringsendMessage(){
rabbitTemplate.convertAndSend("simple-queue","落霞与孤鹜齐飞，秋水共长天一色");return"成功";}}
消息消费方
在SpringBoot中即可使用注解@RabbitListener进行消息的投递，也可使用原生的方式进行消息投递
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.simple;importorg.springframework.amqp.rabbit.annotation.Queue;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassConsumer{/**
* 1、@RabbitListener(queues = {"topic-queue-one"} ),一个@RabbitListener就是一个消费者
* 2、使用queues，如果队列不存在就会报异常，使用queuesToDeclare，如果队列不存在就创建队列
* 3、注意：SpringBoot提供了一个很好的消息确认机制，如果消费方消费的过程中有异常，一定要像service层一样抛出异常不能捕捉
* 4、监听队列，如果有消息就进行消费
* 5、简单模型和工作模型的queue都是绑定在默认交换机上
*/@RabbitListener(queuesToDeclare ={@Queue(name ="simple-queue",durable ="true")})publicvoidgetMessage(String message){System.out.println("springBoot+rabbitMQ: "+message);}}
二、Work模型 ============================================================
特点：工作模型只是在简单模型的基础上增加了多个消费者
消息投递方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.work;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RequestMethod;importorg.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/rabbit")publicclassProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;@RequestMapping(value ="/work",method =RequestMethod.GET)publicvoidsendMessage(){
rabbitTemplate.convertAndSend("work-queue-one","祖国您好！");}}
消息消费方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.work;importorg.springframework.amqp.rabbit.annotation.Queue;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassConsumer{/**
* 1、@RabbitListener(queues = {"topic-queue-one"} ),一个@RabbitListener就是一个消费者
* 2、使用queues，如果队列不存在就会报异常，使用queuesToDeclare，如果队列不存在就创建队列
* 3、注意：SpringBoot提供了一个很好的消息确认机制，如果消费方消费的过程中有异常，一定要像service层一样抛出异常不能捕捉
* 4、监听队列，如果有消息就进行消费
* 5、简单模型和工作模型的queue都是绑定在默认交换机上,工作模型就是在简单模型的基础上增加了多个消费者
*/@RabbitListener(queuesToDeclare =@Queue("work-queue-one"))publicvoidgetMessageOne(String message){System.out.println(message);}@RabbitListener(queuesToDeclare =@Queue("work-queue-two"))publicvoidgetMessageTwo(String message){System.out.println(message);}}
三、Fanout模型 ============================================================
特点：凡是绑定在该交换机上的队列都会收到消息，无关路由键
消息投递方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.fanout;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("rabbit")publicclassProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;@RequestMapping("fanout")//fanout类型的交换机无关路由键publicvoidsendMessage(){
rabbitTemplate.convertAndSend("fanout-exchange","","这是fanout类型的交换机");}}
消息消费方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.fanout;importorg.springframework.amqp.core.ExchangeTypes;importorg.springframework.amqp.rabbit.annotation.Exchange;importorg.springframework.amqp.rabbit.annotation.Queue;importorg.springframework.amqp.rabbit.annotation.QueueBinding;importorg.springframework.amqp.rabbit.annotation.RabbitListener;publicclassConsumer{@RabbitListener(bindings ={@QueueBinding(
value =@Queue(name ="fanout-queue-one"),
exchange =@Exchange(name ="fanout-exchange",type =ExchangeTypes.FANOUT),
key ="fanout-bindingKey-one")})publicvoidgetMessageOne(String message){System.out.println(message);}@RabbitListener(bindings =@QueueBinding(//默认是持久化的
value =@Queue("fanout-queue-two"),
exchange =@Exchange(name ="fanout-exchange",type =ExchangeTypes.FANOUT),//fanout模式下可以不指定key
key ="fanout-bindingKey-two"))publicvoidgetMessageTwo(String message){System.out.println(message);}}
四、Direct模型 ============================================================
特点：根据RoutingKey向对应的队列投递消息
消息投递方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.direct;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/rabbit")publicclassProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;@RequestMapping("/direct")publicvoidsendMessage(){
rabbitTemplate.convertAndSend("direct-exchange","direct-bindingKey-one","五十六个民族，五十六只花");}}
消息消费方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.direct;importorg.springframework.amqp.core.ExchangeTypes;importorg.springframework.amqp.rabbit.annotation.Exchange;importorg.springframework.amqp.rabbit.annotation.Queue;importorg.springframework.amqp.rabbit.annotation.QueueBinding;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassConsumer{@RabbitListener(bindings ={@QueueBinding(value =@Queue(name ="direct-queue-one"),exchange =@Exchange(name ="direct-exchange",type =ExchangeTypes.DIRECT),key ="direct-bindingKey-one")})publicvoidgetMessageOne(String message){System.out.println(message);}@RabbitListener(bindings =@QueueBinding(value =@Queue(name ="direct-queue-two"),exchange =@Exchange(name ="direct-exchange",type =ExchangeTypes.DIRECT),key ="direct-bindingKey-two"))publicvoidgetMessageTwo(String message){System.out.println(message);}}
五、Topic模型 ============================================================
特点：根据RoutingKey向匹配的队列投递消息，其BindingKey中含有#号和号，#匹配多个单词，号匹配一个单词
消息投递方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.topic;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/rabbit")publicclassProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;@RequestMapping("/topic")publicvoidsendMessage(){
rabbitTemplate.convertAndSend("topic-exchange","topic-bindingKey-one.java","行路难，行路难，多歧路，今安在，长风破浪会有时，直挂云帆济沧海");}}
消息消费方
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.topic;importorg.springframework.amqp.core.ExchangeTypes;importorg.springframework.amqp.rabbit.annotation.Exchange;importorg.springframework.amqp.rabbit.annotation.Queue;importorg.springframework.amqp.rabbit.annotation.QueueBinding;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassConsumer{@RabbitListener(bindings =@QueueBinding(
value =@Queue(name ="topic-queue-one"),
exchange =@Exchange(name ="topic-exchange",type =ExchangeTypes.TOPIC),
key ="topic-bindingKey-one"))publicvoidgetMessageOne(String message){System.out.println(message);}@RabbitListener(bindings =@QueueBinding(
value =@Queue(name ="topic-queue-two"),
exchange =@Exchange(name ="topic-exchange",type =ExchangeTypes.TOPIC),
key ="topic-bindingKey-one.*"))publicvoidgetMessageTwo(String message){System.out.println(message);}@RabbitListener(bindings =@QueueBinding(
value =@Queue(name ="topic-queue-three"),
exchange =@Exchange(name ="topic-exchange",type =ExchangeTypes.TOPIC),
key ="topic-bindingKey-one.#"))publicvoidgetMessageThree(String message){System.out.println(message);}}
消息确认机制
应用向MQ投递消息后，消息是否顺利的到达MQ，应用是不知情的，在这一个过程中也无法判断消息是否丢失，RabbitMQ的消息确认机制本质就是为了保证消息的消息的不丢失。
RabbitMQ的消息确认机制包含两部分：
MQ收到消息的确认
和
消费方消费消息的确认
MQ收到消息的确认
MQ收到消息的确认包含两部分：
消息到达交换机的确认
和
消息到达队列的确认
，所谓的确认其实就是在消息到达交换机或队列后，告知应用交换机或队列收到消息。
原生的确认方式：
packagecom.example.demo.rabbitMQ.confirm;importcom.example.demo.rabbitMQ.RabbitMQUtils;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassRabbitMQConfirm{publicstaticvoidmain(String[] args)throwsException{Connection connection =RabbitMQUtils.getConnection();Channel channel = connection.createChannel();
channel.queueDeclare("simple-queue",true,false,false,null);//开启交换机的消息确认机制=>消息是否到达交换机的确认
channel.confirmSelect();
channel.addConfirmListener(newConfirmListener(){//消息到达交换机的回调@OverridepublicvoidhandleAck(long l,boolean b)throwsIOException{System.out.println("交换机成功接收到消息！");}//消息未到达交换机的回调@OverridepublicvoidhandleNack(long l,boolean b)throwsIOException{System.out.println("交换机未接收到消息！");}});
channel.addReturnListener(newReturnCallback(){//消息未到达队列的回调@Overridepublicvoidhandle(Return aReturn){System.out.println("队列未接收到消息！");}});/**
* 第一个参数：交换机名称
* 第二个参数：RoutingKey==BindingKey，空串代表默认交换机，在简单模式中，BindingKey就是队列名称
* 第三个参数：是否开启队列的消息确认机制=>消息是否到达队列的确认
* 第四个参数：消息属性（消息存活时间，消息是否持久化）
* 第五个参数：消息内容
*/
channel.basicPublish("","simple-queue",true,null,"海客谈瀛洲，烟涛注微茫信难求；越人语天姥，云霞明灭或可睹。天姥连天向天横，势拔注五岳掩赤城".getBytes());}}
如何保证消息的可靠性：
假设：现在有一个支付流水系统和一个积分系统，支付流水系统作为上游业务，每产生一条支付流水记录，积分系统作为下游业务就需要增加一条对应的积分记录。
思考：由于该事物是分布在两个不同的系统中，所以就不能采用事物回滚的方案，
事物回滚的前提是必须是同一个连接，且异常只能抛出不能解决
。
1、方案一的问题：支付流水系统先记录支付流水信息，再向MQ中投递一条消息，如果本地记录成功，消息投递失败，那么支付流水系统中的记录和积分系统中的记录就会不一致。
2、方案二的问题：先向MQ中先投递消息，支付流水系统再记录支付流水信息，如果消息投递成功，本地记录失败，那么支付流水系统中的记录和积分系统中的记录就会不一致。
终极解决方案：
1、支付流水系统记录流水信息的同时，将要投递到MQ的消息一并记录(二者需要在同一个事务中)
2、如果消息投递成功，删除本地记录的要投递到MQ的消息，如果投递失败，引入消息补偿系统，后续通过消息补偿系统读取本地记录的要投递到MQ的消息，将读取到的消息继续向MQ中投递，一旦投递成功就删除本地记录的要投递到MQ的消息，如果投递失败次数超过5次，就需要人工手动干预
SpringBoot+RabbitMQ的消息确认机制：
packagecom.example.demo.config.rabbitMQConfig;importorg.springframework.amqp.core.*;importorg.springframework.amqp.rabbit.connection.ConnectionFactory;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitAdmin;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.context.annotation.Bean;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Component;importjavax.annotation.PostConstruct;importjava.util.concurrent.TimeUnit;@ComponentpublicclassRabbitMQConfig{/**
* springBoot默认注入的RabbitTempleate是没有实现消息确认机制的，需要我们手动实现
*/privateRabbitAdmin rabbitAdmin;privateConnectionFactory connectionFactory;privateStringRedisTemplate stringRedisTemplate;publicRabbitMQConfig(ConnectionFactory connectionFactory,StringRedisTemplate stringRedisTemplate){this.connectionFactory = connectionFactory;this.stringRedisTemplate = stringRedisTemplate;this.rabbitAdmin =newRabbitAdmin(connectionFactory);}@PostConstructpublicvoidinitExchangeAndQueue(){/*====================================================方式一====================================================*///定义队列的相关信息Queue queue =QueueBuilder.durable("rabbitMQ-springBoot-queue").build();//这一步才是真正的在RabbitMQ中创建队列
rabbitAdmin.declareQueue(queue);//定义交换机的相关信息Exchange exchange =ExchangeBuilder.directExchange("rabbitMQ-springBoot-exchange").build();//这一步才是真正的在RabbitMQ中创建交换机
rabbitAdmin.declareExchange(exchange);Binding binding =BindingBuilder.bind(queue).to(exchange).with("rabbitMQ-springBoot-key").noargs();//这一步才是真正的将队列绑定到交换机上
rabbitAdmin.declareBinding(binding);/*====================================================方式二====================================================*/
rabbitAdmin.declareBinding(BindingBuilder.bind(newQueue("rabbitMQ-springBoot-queue",true)).to(newDirectExchange("rabbitMQ-springBoot-exchange",true,false)).with("rabbitMQ-springBoot-key"));}@Bean("appRabbitTempleate")publicRabbitTemplategetRabbitTempleate(){RabbitTemplate rabbitTemplate =newRabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);//开启RabbitMQ的消息到达队列的消息确认机制
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(newRabbitTemplate.ConfirmCallback(){@Overridepublicvoidconfirm(CorrelationData correlationData,boolean isReceive,String s){String messageId = correlationData.getId();if(isReceive){//交换机成功接收到消息的逻辑boolean isExit = stringRedisTemplate.opsForSet().isMember("ORDER#MESSAGE#SEND#FAILD",messageId);//通过redis中是否存在messageId判断消息是否成功到达队列if(!isExit){//如果messageId不存在，说明消息成功到达队列==》当下情况：消息既到达交换机又成功到达了队列==》消息投递成功/**
* 消息投递成功的逻辑：
* 删除本地记录表中的记录
*/}else{//如果messageId不存在，说明消息没有到达队列==》当下情况：消息到达交换机并未到达队列/**
* 消息未到达队列，消息投递失败的逻辑
* 需要移除redis中对应的messageId
*/
stringRedisTemplate.opsForSet().remove("ORDER#MESSAGE#SEND#FAILD",messageId);}}else{//交换机未接收到消息的逻辑/**
* 消息未到达交换机，消息投递失败的逻辑
* 需要移除redis中对应的messageId[如果交换机不存在，消息未到达队列的逻辑不会走，redis中就不会有对应的messageId；如果交换机存在，消息未到达队列的逻辑就会走，redis中就会有对应的messageId；不管redis中有没有，保险起见，都需要做移除操作]
*/
stringRedisTemplate.opsForSet().remove("ORDER#MESSAGE#SEND#FAILD",messageId);}}});
rabbitTemplate.setReturnCallback(newRabbitTemplate.ReturnCallback(){@OverridepublicvoidreturnedMessage(Message message,int i,String s,String s1,String s2){/**
* 1、此处如果执行了，说明消息未到达队列，消息投递失败
* 2、此处如果没有执行，分两种情况：
* A、没有执行，并不能说明消息一定投递成功了，因为没有交换机该逻辑也不会执行，消息投递失败
* B、没有执行，投递成功了，消息成功到达了队列
* 该逻辑的执行优先于交换机的逻辑执行
*/String messageId = message.getMessageProperties().getMessageId();//将投递失败的消息的唯一标识存入redis中
stringRedisTemplate.opsForSet().add("ORDER#MESSAGE#SEND#FAILD",messageId);}});return rabbitTemplate;}}
消息发送方：
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.confirm;importcom.alibaba.fastjson.JSONObject;importcom.example.demo.model.MessageRecordModel;importcom.example.demo.model.OrderModel;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.core.MessageDeliveryMode;importorg.springframework.amqp.core.MessageProperties;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.web.bind.annotation.RequestMapping;importorg.springframework.web.bind.annotation.RequestMethod;importorg.springframework.web.bind.annotation.RestController;importjavax.annotation.Resource;importjava.nio.charset.StandardCharsets;importjava.util.UUID;importjava.util.concurrent.atomic.AtomicInteger;@RestController@RequestMapping("/rabbit")publicclassRabbitMQConfirm{@Resource(name ="appRabbitTempleate")privateRabbitTemplate rabbitTemplate;privateAtomicInteger atomicInteger =newAtomicInteger(10000);@RequestMapping(value ="/confirm",method =RequestMethod.GET)publicvoidsendMessage(){//这是订单逻辑==================================================OrderModel orderModel=newOrderModel();String uniqueId =System.currentTimeMillis()+"#"+UUID.randomUUID();
orderModel.setOrderId(uniqueId);String messageContent =JSONObject.toJSONString(orderModel);//这是记录逻辑==================================================MessageRecordModel messageRecordModel =newMessageRecordModel();
messageRecordModel.setExchange("");
messageRecordModel.setRoutionKey("");
messageRecordModel.setMessageContent(messageContent);//这是向rabbit投递消息的和逻辑==================================================CorrelationData correlationData =newCorrelationData();
correlationData.setId(uniqueId);MessageProperties messageProperties =newMessageProperties();//设置消息的持久化
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
messageProperties.setMessageId(uniqueId);Message message =newMessage(messageContent.getBytes(StandardCharsets.UTF_8),messageProperties);
rabbitTemplate.send(messageRecordModel.getExchange(),messageRecordModel.getRoutionKey(),message,correlationData);}}
消息消费：
packagecom.example.demo.rabbitMQ.springBootrabbitMQ.comprehensive.consumer;importcom.alibaba.fastjson.JSONObject;importcom.example.demo.model.OrderModel;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importorg.redisson.api.RLock;importorg.redisson.api.RedissonClient;importorg.springframework.amqp.core.ExchangeTypes;importorg.springframework.amqp.rabbit.annotation.Exchange;importorg.springframework.amqp.rabbit.annotation.Queue;importorg.springframework.amqp.rabbit.annotation.QueueBinding;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.amqp.support.AmqpHeaders;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.messaging.handler.annotation.Header;importorg.springframework.messaging.handler.annotation.Payload;importorg.springframework.stereotype.Component;importjava.io.IOException;@ComponentpublicclassConsumer{@AutowiredprivateStringRedisTemplate stringRedisTemplate;@AutowiredprivateRedissonClient redissonClient;@RabbitListener(bindings =@QueueBinding(value =@Queue(value ="order_queue",durable ="true"),
exchange =@Exchange(value ="order_exchange",type =ExchangeTypes.DIRECT),
key ="order_bindingKey"),
ackMode ="MANUAL")/**
* // 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
* RECORD,
* // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
* BATCH,
* // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
* TIME,
* // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
* COUNT,
* // TIME | COUNT 有一个条件满足时提交
* COUNT_TIME,
* // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
* MANUAL,
* // 手动调用Acknowledgment.acknowledge()后立即提交
* MANUAL_IMMEDIATE,
*
*/publicvoidgetMessage(@PayloadString message,@Header(AmqpHeaders.DELIVERY_TAG)long deliveryTag,Channel channel)throwsIOException{
channel.basicQos(1);OrderModel messageContent =JSONObject.parseObject(message,OrderModel.class);String orderId = messageContent.getOrderId();RLock lock = redissonClient.getLock("ZFLS#"+orderId);try{if(stringRedisTemplate.opsForValue().get("ZFLS#"+orderId)!=null){
lock.lock();if(stringRedisTemplate.opsForValue().get("ZFLS#"+orderId)!=null){//获取到消息后开始进行记分处理==================================================/**
* 此处代码省略
*///积分处理完成后确认消息已被处理ACK==================================================
channel.basicAck(deliveryTag,false);//为了防止消息重复消费，在消息处理完成后还需将消息放入redis中
stringRedisTemplate.opsForValue().set("ZFLS#"+orderId,"true");}}}finally{
lock.unlock();}}}
消息重投
//使用定时器或者quartz进行消息的重投
面试题：MQ消息投递如何确保消息的可靠性（即：在投递的过程中消息不会丢失，在消费的过程中消息不会被重复消费）？
首先开启交换机的消息确认机制，再开启队列的消息确认机制，最后消费方采用手动ACK的方式替代消息自动确认机制。
https://blog.csdn.net/weixin_42039228/article/details/123493937
https://blog.csdn.net/Qynwang/article/details/130490528
本文转载自: https://blog.csdn.net/qq_44861126/article/details/130283245
版权归原作者 小逗比iswho 所有， 如有侵权，请联系我们删除。
版权归原作者 小逗比iswho 所有， 如有侵权，请联系我们删除。