RabbitMQ快速使用代码手册

本篇博客的内容为RabbitMQ在开发过程中的快速上手使用，侧重于代码部分，几乎没有相关概念的介绍，相关概念请参考以下csdn博客，两篇都是我找的精华帖，供大家学习。本篇博客也持续更新~~~
内容代码部分由于word转md格式有些问题，可以直接查看我的有道云笔记，链接：https://note.youdao.com/s/Ab7Cjiu

参考文档

csdn博客：

基础部分：https://blog.csdn.net/qq_35387940/article/details/100514134

高级部分：https://blog.csdn.net/weixin_49076273/article/details/124991012

application.yml

server:

port: 8021

spring:

#给项目来个名字

application:

name: rabbitmq-provider

#配置rabbitMq 服务器

rabbitmq:

host: 127.0.0.1

port: 5672

username: root

password: root

#虚拟host 可以不设置,使用server默认host

virtual-host: JCcccHost

#确认消息已发送到交换机(Exchange)

#publisher-confirms: true

publisher-confirm-type: correlated

#确认消息已发送到队列(Queue)

publisher-returns: true

完善更多信息

spring:

rabbitmq:

host: localhost

port: 5672

virtual-host: /

username: guest

password: guest

publisher-confirm-type: correlated

publisher-returns: true

template:

mandatory: true

retry:

#发布重试，默认false

enabled: true

#重试时间 默认1000ms

initial-interval: 1000

#重试最大次数 最大3

max-attempts: 3

#重试最大间隔时间

max-interval: 10000

#重试的时间隔乘数，比如配2，0
第一次等于10s，第二次等于20s，第三次等于40s

multiplier: 1

listener:

\# 默认配置是simple

type: simple

simple:

\# 手动ack Acknowledge mode of container. auto none

acknowledge-mode: manual

#消费者调用程序线程的最小数量

concurrency: 10

#消费者最大数量

max-concurrency: 10

#限制消费者每次只处理一条信息，处理完在继续下一条

prefetch: 1

#启动时是否默认启动容器

auto-startup: true

#被拒绝时重新进入队列

default-requeue-rejected: true

相关注解说明

@RabbitListener 注解是指定某方法作为消息消费的方法，例如监听某 Queue
里面的消息。

@RabbitListener标注在方法上，直接监听指定的队列，此时接收的参数需要与发送市类型一致。

\@Component

public class PointConsumer {

//监听的队列名

\@RabbitListener(queues = \"point.to.point\")

public void processOne(String name) {

System.out.println(\"point.to.point：\" + name);

}

}

@RabbitListener 可以标注在类上面，需配合 @RabbitHandler 注解一起使用

@RabbitListener 标注在类上面表示当有收到消息的时候，就交给
@RabbitHandler 的方法处理，根据接受的参数类型进入具体的方法中。

\@Component

\@RabbitListener(queues = \"consumer_queue\")

public class Receiver {

\@RabbitHandler

public void processMessage1(String message) {

System.out.println(message);

}

\@RabbitHandler

public void processMessage2(byte\[\] message) {

System.out.println(new String(message));

}

}

@Payload

可以获取消息中的 body 信息

\@RabbitListener(queues = \"debug\")

public void processMessage1(@Payload String body) {

System.out.println(\"body：\"+body);

}

@Header，@Headers

可以获得消息中的 headers 信息

\@RabbitListener(queues = \"debug\")

public void processMessage1(@Payload String body, \@Header String token)
{

System.out.println(\"body：\"+body);

System.out.println(\"token：\"+token);

}

\@RabbitListener(queues = \"debug\")

public void processMessage1(@Payload String body, \@Headers
Map\<String,Object\> headers) {

System.out.println(\"body：\"+body);

System.out.println(\"Headers：\"+headers);

}

快速使用

配置xml文件

<dependency\>

\<groupId\>org.springframework.boot\</groupId\>

\<artifactId\>spring-boot-starter-amqp\</artifactId\>

\</dependency\>

配置exchange、queue

注解快速创建版本

\@Configuration

public class RabbitmqConfig {

//创建交换机

//通过ExchangeBuilder能创建direct、topic、Fanout类型的交换机

\@Bean(\"bootExchange\")

public Exchange bootExchange() {

return
ExchangeBuilder.topicExchange(\"zx_topic_exchange\").durable(true).build();

}

//创建队列

\@Bean(\"bootQueue\")

public Queue bootQueue() {

return QueueBuilder.durable(\"zx_queue\").build();

}

/\*\*

\* 将队列与交换机绑定

\*

\* \@param queue

\* \@param exchange

\* \@return

\*/

\@Bean

public Binding bindQueueExchange(@Qualifier(\"bootQueue\") Queue queue,
\@Qualifier(\"bootExchange\") Exchange exchange) {

return
BindingBuilder.bind(queue).to(exchange).with(\"boot.#\").noargs();

}

}

Direct

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.DirectExchange;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/\*\*

\* \@Author : JCccc

\* \@CreateTime : 2019/9/3

\* \@Description :

\*\*/

\@Configuration

public class DirectRabbitConfig {

//队列 起名：TestDirectQueue

\@Bean

public Queue TestDirectQueue() {

//
durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效

//
exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable

//
autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。

// return new Queue(\"TestDirectQueue\",true,true,false);

//一般设置一下队列的持久化就好,其余两个就是默认false

return new Queue(\"TestDirectQueue\",true);

}

//Direct交换机 起名：TestDirectExchange

\@Bean

DirectExchange TestDirectExchange() {

// return new DirectExchange(\"TestDirectExchange\",true,true);

return new DirectExchange(\"TestDirectExchange\",true,false);

}

//绑定 将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting

\@Bean

Binding bindingDirect() {

return
BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(\"TestDirectRouting\");

}

\@Bean

DirectExchange lonelyDirectExchange() {

return new DirectExchange(\"lonelyDirectExchange\");

}

}

Fanout

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.FanoutExchange;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/\*\*

\* \@Author : JCccc

\* \@CreateTime : 2019/9/3

\* \@Description :

\*\*/

\@Configuration

public class FanoutRabbitConfig {

/\*\*

\* 创建三个队列 ：fanout.A fanout.B fanout.C

\* 将三个队列都绑定在交换机 fanoutExchange 上

\* 因为是扇型交换机, 路由键无需配置,配置也不起作用

\*/

\@Bean

public Queue queueA() {

return new Queue(\"fanout.A\");

}

\@Bean

public Queue queueB() {

return new Queue(\"fanout.B\");

}

\@Bean

public Queue queueC() {

return new Queue(\"fanout.C\");

}

\@Bean

FanoutExchange fanoutExchange() {

return new FanoutExchange(\"fanoutExchange\");

}

\@Bean

Binding bindingExchangeA() {

return BindingBuilder.bind(queueA()).to(fanoutExchange());

}

\@Bean

Binding bindingExchangeB() {

return BindingBuilder.bind(queueB()).to(fanoutExchange());

}

\@Bean

Binding bindingExchangeC() {

return BindingBuilder.bind(queueC()).to(fanoutExchange());

}

}

Topic

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.core.TopicExchange;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/\*\*

\* \@Author : JCccc

\* \@CreateTime : 2019/9/3

\* \@Description :

\*\*/

\@Configuration

public class TopicRabbitConfig {

//绑定键

public final static String man = \"topic.man\";

public final static String woman = \"topic.woman\";

\@Bean

public Queue firstQueue() {

return new Queue(TopicRabbitConfig.man);

}

\@Bean

public Queue secondQueue() {

return new Queue(TopicRabbitConfig.woman);

}

\@Bean

TopicExchange exchange() {

return new TopicExchange(\"topicExchange\");

}

//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man

//这样只要是消息携带的路由键是topic.man,才会分发到该队列

\@Bean

Binding bindingExchangeMessage() {

return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);

}

//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#

// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列

\@Bean

Binding bindingExchangeMessage2() {

return
BindingBuilder.bind(secondQueue()).to(exchange()).with(\"topic.#\");

}

}

生产者发送消息

直接发送给队列

//指定消息队列的名字，直接发送消息到消息队列中

\@Test

public void testSimpleQueue() {

// 队列名称

String queueName = \"simple.queue\";

// 消息

String message = \"hello, spring amqp!\";

// 发送消息

rabbitTemplate.convertAndSend(queueName, message);

}

发送给交换机，然后走不同的模式

////指定交换机的名字，将消息发送给交换机，然后不同模式下，消息队列根据key得到消息

\@Test

public void testSendDirectExchange() {

// 交换机名称，有三种类型

String exchangeName = \"itcast.direct\";

// 消息

String message =
\"红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！\";

// 发送消息，red为队列的key，因此此队列会得到消息

rabbitTemplate.convertAndSend(exchangeName, \"red\", message);

}

也可以将发送的消息封装到HashMap中然后发送给交换机

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.HashMap;

import java.util.Map;

import java.util.UUID;

/\*\*

\* \@Author : JCccc

\* \@CreateTime : 2019/9/3

\* \@Description :

\*\*/

\@RestController

public class SendMessageController {

\@Autowired

RabbitTemplate rabbitTemplate;
//使用RabbitTemplate,这提供了接收/发送等等方法

\@GetMapping(\"/sendDirectMessage\")

public String sendDirectMessage() {

String messageId = String.valueOf(UUID.randomUUID());

String messageData = \"test message, hello!\";

String createTime =
LocalDateTime.now().format(DateTimeFormatter.ofPattern(\"yyyy-MM-dd
HH:mm:ss\"));

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

map.put(\"messageId\",messageId);

map.put(\"messageData\",messageData);

map.put(\"createTime\",createTime);

//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange

rabbitTemplate.convertAndSend(\"TestDirectExchange\",
\"TestDirectRouting\", map);

return \"ok\";

}

}

消费者接收消息

//使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。

\@Component

public class MessageListener {

\@RabbitListener(queues = \"direct_queue\")

public void receive(String id){

System.out.println(\"已完成短信发送业务(rabbitmq direct)，id：\"+id);

}

}

参数用Map接收也可以

\@Component

\@RabbitListener(queues = \"TestDirectQueue\")//监听的队列名称
TestDirectQueue

public class DirectReceiver {

\@RabbitHandler

public void process(Map testMessage) {

System.out.println(\"DirectReceiver消费者收到消息 : \" +
testMessage.toString());

}

}

高级特性

消息可靠性传递

有confirm和return两种

在application.yml中添加以下配置项：

server:

port: 8021

spring:

#给项目来个名字

application:

name: rabbitmq-provider

#配置rabbitMq 服务器

rabbitmq:

host: 127.0.0.1

port: 5672

username: root

password: root

#虚拟host 可以不设置,使用server默认host

virtual-host: JCcccHost

#确认消息已发送到交换机(Exchange)

#publisher-confirms: true

publisher-confirm-type: correlated

#确认消息已发送到队列(Queue)

publisher-returns: true

有两种配置方法：

写到配置类中

写到工具类或者普通类中，但是这个类得实现那两个接口

写法一

编写消息确认回调函数

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import org.springframework.amqp.rabbit.connection.CorrelationData;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

\@Configuration

public class RabbitConfig {

\@Bean

public RabbitTemplate createRabbitTemplate(ConnectionFactory
connectionFactory){

RabbitTemplate rabbitTemplate = new RabbitTemplate();

rabbitTemplate.setConnectionFactory(connectionFactory);

//设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数

rabbitTemplate.setMandatory(true);

rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

\@Override

public void confirm(CorrelationData correlationData, boolean ack, String
cause) {

System.out.println(\"ConfirmCallback:
\"+\"相关数据：\"+correlationData);

System.out.println(\"ConfirmCallback: \"+\"确认情况：\"+ack);

System.out.println(\"ConfirmCallback: \"+\"原因：\"+cause);

}

});

rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {

\@Override

public void returnedMessage(Message message, int replyCode, String
replyText, String exchange, String routingKey) {

System.out.println(\"ReturnCallback: \"+\"消息：\"+message);

System.out.println(\"ReturnCallback: \"+\"回应码：\"+replyCode);

System.out.println(\"ReturnCallback: \"+\"回应信息：\"+replyText);

System.out.println(\"ReturnCallback: \"+\"交换机：\"+exchange);

System.out.println(\"ReturnCallback: \"+\"路由键：\"+routingKey);

}

});

return rabbitTemplate;

}

}

写法二

\@Component

\@Slf4j

public class SmsRabbitMqUtils implements RabbitTemplate.ConfirmCallback,
RabbitTemplate.ReturnsCallback {

\@Resource

private RedisTemplate\<String, String\> redisTemplate;

\@Resource

private RabbitTemplate rabbitTemplate;

private String finalId = null;

private SmsDTO smsDTO = null;

/\*\*

\* 发布者确认的回调

\*

\* \@param correlationData 回调的相关数据。

\* \@param b ack为真，nack为假

\* \@param s 一个可选的原因，用于nack，如果可用，否则为空。

\*/

\@Override

public void confirm(CorrelationData correlationData, boolean b, String
s) {

// 消息发送成功，将redis中消息的状态（status）修改为1

if (b) {

redisTemplate.opsForHash().put(RedisConstant.SMS_MESSAGE_PREFIX +
finalId, \"status\", 1);

} else {

// 发送失败，放入redis失败集合中，并删除集合数据

log.error(\"短信消息投送失败：{}\--\>{}\", correlationData, s);

redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId);

redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId,
this.smsDTO);

}

}

/\*\*

\* 发生异常时的消息返回提醒

\*

\* \@param returnedMessage

\*/

\@Override

public void returnedMessage(ReturnedMessage returnedMessage) {

log.error(\"发生异常，返回消息回调:{}\", returnedMessage);

// 发送失败，放入redis失败集合中，并删除集合数据

redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId);

redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId,
this.smsDTO);

}

\@PostConstruct

public void init() {

rabbitTemplate.setConfirmCallback(this);

rabbitTemplate.setReturnsCallback(this);

}

}

消息确认机制

手动确认

yml配置

#手动确认 manual

listener:

simple:

acknowledge-mode: manual

写法一

首先在消费者项目中创建MessageListenerConfig

import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver;

import org.springframework.amqp.core.AcknowledgeMode;

import org.springframework.amqp.core.Queue;

import
org.springframework.amqp.rabbit.connection.CachingConnectionFactory;

import
org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

\@Configuration

public class MessageListenerConfig {

\@Autowired

private CachingConnectionFactory connectionFactory;

\@Autowired

private MyAckReceiver myAckReceiver;//消息接收处理类

\@Bean

public SimpleMessageListenerContainer simpleMessageListenerContainer() {

SimpleMessageListenerContainer container = new
SimpleMessageListenerContainer(connectionFactory);

container.setConcurrentConsumers(1);

container.setMaxConcurrentConsumers(1);

container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //
RabbitMQ默认是自动确认，这里改为手动确认消息

//设置一个队列

container.setQueueNames(\"TestDirectQueue\");

//如果同时设置多个如下： 前提是队列都是必须已经创建存在的

//
container.setQueueNames(\"TestDirectQueue\",\"TestDirectQueue2\",\"TestDirectQueue3\");

//另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues

//container.setQueues(new Queue(\"TestDirectQueue\",true));

//container.addQueues(new Queue(\"TestDirectQueue2\",true));

//container.addQueues(new Queue(\"TestDirectQueue3\",true));

container.setMessageListener(myAckReceiver);

return container;

}

}

然后创建手动确认监听类MyAckReceiver(手动确认模式需要实现ChannelAwareMessageListener)

import com.rabbitmq.client.Channel;

import org.springframework.amqp.core.Message;

import
org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;

import java.io.ObjectInputStream;

import java.util.Map;

\@Component

public class MyAckReceiver implements ChannelAwareMessageListener {

\@Override

public void onMessage(Message message, Channel channel) throws Exception
{

long deliveryTag = message.getMessageProperties().getDeliveryTag();

try {

byte\[\] body = message.getBody();

ObjectInputStream ois = new ObjectInputStream(new
ByteArrayInputStream(body));

Map\<String,String\> msgMap = (Map\<String,String\>) ois.readObject();

String messageId = msgMap.get(\"messageId\");

String messageData = msgMap.get(\"messageData\");

String createTime = msgMap.get(\"createTime\");

ois.close();

System.out.println(\" MyAckReceiver messageId:\"+messageId+\"
messageData:\"+messageData+\" createTime:\"+createTime);

System.out.println(\"消费的主题消息来自：\"+message.getMessageProperties().getConsumerQueue());

channel.basicAck(deliveryTag, true);
//第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认
delivery_tag 小于等于传入值的所有消息

//channel.basicReject(deliveryTag,
true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝

} catch (Exception e) {

channel.basicReject(deliveryTag, false);

e.printStackTrace();

}

}

}

如果想实现不同的队列，有不同的监听确认处理机制，做不同的业务处理，那么这样做：

首先需要在配置类中绑定队列，然后只需要根据消息来自不同的队列名进行区分处理即可

import com.rabbitmq.client.Channel;

import org.springframework.amqp.core.Message;

import
org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;

import java.io.ObjectInputStream;

import java.util.Map;

\@Component

public class MyAckReceiver implements ChannelAwareMessageListener {

\@Override

public void onMessage(Message message, Channel channel) throws Exception
{

long deliveryTag = message.getMessageProperties().getDeliveryTag();

try {

byte\[\] body = message.getBody();

ObjectInputStream ois = new ObjectInputStream(new
ByteArrayInputStream(body));

Map\<String,String\> msgMap = (Map\<String,String\>) ois.readObject();

String messageId = msgMap.get(\"messageId\");

String messageData = msgMap.get(\"messageData\");

String createTime = msgMap.get(\"createTime\");

ois.close();

if
(\"TestDirectQueue\".equals(message.getMessageProperties().getConsumerQueue())){

System.out.println(\"消费的消息来自的队列名为：\"+message.getMessageProperties().getConsumerQueue());

System.out.println(\"消息成功消费到 messageId:\"+messageId+\"
messageData:\"+messageData+\" createTime:\"+createTime);

System.out.println(\"执行TestDirectQueue中的消息的业务处理流程\...\...\");

}

if
(\"fanout.A\".equals(message.getMessageProperties().getConsumerQueue())){

System.out.println(\"消费的消息来自的队列名为：\"+message.getMessageProperties().getConsumerQueue());

System.out.println(\"消息成功消费到 messageId:\"+messageId+\"
messageData:\"+messageData+\" createTime:\"+createTime);

System.out.println(\"执行fanout.A中的消息的业务处理流程\...\...\");

}

channel.basicAck(deliveryTag, true);

//channel.basicReject(deliveryTag, true);//为true会重新放回队列

} catch (Exception e) {

channel.basicReject(deliveryTag, false);

e.printStackTrace();

}

}

}

写法二

\@Component

\@Slf4j

public class SendSmsListener {

\@Resource

private RedisTemplate\<String, String\> redisTemplate;

\@Resource

private SendSmsUtils sendSmsUtils;

/\*\*

\* 监听发送短信普通队列

\* \@param smsDTO

\* \@param message

\* \@param channel

\* \@throws IOException

\*/

\@RabbitListener(queues = SMS_QUEUE_NAME)

public void sendSmsListener(SmsDTO smsDTO, Message message, Channel
channel) throws IOException {

String messageId = message.getMessageProperties().getMessageId();

int retryCount = (int)
redisTemplate.opsForHash().get(RedisConstant.SMS_MESSAGE_PREFIX +
messageId, \"retryCount\");

if (retryCount \> 3) {

//重试次数大于3，直接放到死信队列

log.error(\"短信消息重试超过3次：{}\", messageId);

//basicReject方法拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。

//该方法reject后，该消费者还是会消费到该条被reject的消息。

channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);

redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + messageId);

return;

}

try {

String phoneNum = smsDTO.getPhoneNum();

String code = smsDTO.getCode();

if(StringUtils.isAnyBlank(phoneNum,code)){

throw new RuntimeException(\"sendSmsListener参数为空\");

}

// 发送消息

SendSmsResponse sendSmsResponse = sendSmsUtils.sendSmsResponse(phoneNum,
code);

SendStatus\[\] sendStatusSet = sendSmsResponse.getSendStatusSet();

SendStatus sendStatus = sendStatusSet\[0\];

if(!\"Ok\".equals(sendStatus.getCode()) \|\|!\"send
success\".equals(sendStatus.getMessage())){

throw new RuntimeException(\"发送验证码失败\");

}

//手动确认消息

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

log.info(\"短信发送成功：{}\",smsDTO);

redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + messageId);

} catch (Exception e) {

redisTemplate.opsForHash().put(RedisConstant.SMS_MESSAGE_PREFIX+messageId,\"retryCount\",retryCount+1);

channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);

}

}

/\*\*

\* 监听到发送短信死信队列

\* \@param sms

\* \@param message

\* \@param channel

\* \@throws IOException

\*/

\@RabbitListener(queues = SMS_DELAY_QUEUE_NAME)

public void smsDelayQueueListener(SmsDTO sms, Message message, Channel
channel) throws IOException {

try{

log.error(\"监听到死信队列消息==\>{}\",sms);

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

}catch (Exception e){

channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);

}

}

}

消费端限流

#配置RabbitMQ

spring:

rabbitmq:

host: 192.168.126.3

port: 5672

username: guest

password: guest

virtual-host: /

#开启自动确认 none 手动确认 manual

listener:

simple:

#消费端限流机制必须开启手动确认

acknowledge-mode: manual

#消费端最多拉取的消息条数，签收后不满该条数才会继续拉取

prefetch: 5

消息存活时间TTL

可以设置队列的存活时间，也可以设置具体消息的存活时间

设置队列中所有消息的存活时间

return QueueBuilder

.durable(QUEUE_NAME)//队列持久化

.ttl(10000)//设置队列的所有消息存活10s

.build();

即在创建队列时，设置存活时间

设置某条消息的存活时间

//发送消息，并设置该消息的存活时间

\@Test

public void testSendMessage()

{

//1.创建消息属性

MessageProperties messageProperties = new MessageProperties();

//2.设置存活时间

messageProperties.setExpiration(\"10000\");

//3.创建消息对象

Message message = new
Message(\"sendMessage\...\".getBytes(),messageProperties);

//4.发送消息

rabbitTemplate.convertAndSend(\"my_topic_exchange1\",\"my_routing\",message);

}

若设置中间的消息的存活时间，当过期时，该消息不会被移除，但是该消息已经不会被消费了，需要等到该消息到队里顶端才会被移除。因为队列是头出，尾进，故而要移除它需要等到它在顶端时才可以。

在队列设置存活时间，也在单条消息设置存活时间，则以时间短的为准

死信队列

死信队列和普通队列没有任何区别，只需要将普通队列需要绑定死信交换机和死信队列就能够实现功能

import org.springframework.amqp.core.\*;

import org.springframework.beans.factory.annotation.Qualifier;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

\@Configuration//Rabbit配置类

public class RabbitConfig4 {

private final String DEAD_EXCHANGE = \"dead_exchange\";

private final String DEAD_QUEUE = \"dead_queue\";

private final String NORMAL_EXCHANGE = \"normal_exchange\";

private final String NORMAL_QUEUE = \"normal_queue\";

//创建死信交换机

\@Bean(DEAD_EXCHANGE)

public Exchange deadExchange()

{

return ExchangeBuilder

.topicExchange(DEAD_EXCHANGE)//交换机类型 ;参数为名字
topic为通配符模式的交换机

.durable(true)//是否持久化，true即存到磁盘,false只在内存上

.build();

}

//创建死信队列

\@Bean(DEAD_QUEUE)

public Queue deadQueue()

{

return QueueBuilder

.durable(DEAD_QUEUE)//队列持久化

//.maxPriority(10)//设置队列的最大优先级，最大可以设置255，但官网推荐不超过10，太高比较浪费资源

.build();

}

//死信交换机绑定死信队列

\@Bean

//@Qualifier注解,使用名称装配进行使用

public Binding bindDeadQueue(@Qualifier(DEAD_EXCHANGE) Exchange
exchange, \@Qualifier(DEAD_QUEUE) Queue queue)

{

return BindingBuilder

.bind(queue)

.to(exchange)

.with(\"dead_routing\")

.noargs();

}

//创建普通交换机

\@Bean(NORMAL_EXCHANGE)

public Exchange normalExchange()

{

return ExchangeBuilder

.topicExchange(NORMAL_EXCHANGE)//交换机类型 ;参数为名字
topic为通配符模式的交换机

.durable(true)//是否持久化，true即存到磁盘,false只在内存上

.build();

}

//创建普通队列

\@Bean(NORMAL_QUEUE)

public Queue normalQueue()

{

return QueueBuilder

.durable(NORMAL_QUEUE)//队列持久化

//.maxPriority(10)//设置队列的最大优先级，最大可以设置255，但官网推荐不超过10，太高比较浪费资源

.deadLetterExchange(DEAD_EXCHANGE)//绑定死信交换机

.deadLetterRoutingKey(\"dead_routing\")//死信队列路由关键字

.ttl(10000)//消息存活10s

.maxLength(10)//队列最大长度为10

.build();

}

//普通交换机绑定普通队列

\@Bean

//@Qualifier注解,使用名称装配进行使用

public Binding bindNormalQueue(@Qualifier(NORMAL_EXCHANGE) Exchange
exchange, \@Qualifier(NORMAL_QUEUE) Queue queue)

{

return BindingBuilder

.bind(queue)

.to(exchange)

.with(\"my_routing\")

.noargs();

}

}

延迟队列

RabbitMQ并未实现延迟队列功能，所以可以通过死信队列实现延迟队列的功能

即给普通队列设置存活时间30分钟，过期后发送至死信队列，在死信消费者监听死信队列消息，查看订单状态，是否支付，未支付则取消订单，回退库存即可。

消费者监听延迟队列

\@Component

public class ExpireOrderConsumer {

//监听过期订单队列

\@RabbitListener(queues = \"expire_queue\")

public void listenMessage(String orderId)

{

//模拟处理数据库等业务

System.out.println(\"查询\"+orderId+\"号订单的状态，如果已支付无需处理，如果未支付则回退库存\");

}

}

控制层代码

\@RestController

public class OrderController {

\@Autowired

private RabbitTemplate rabbitTemplate;

\@RequestMapping(value = \"/place/{orderId}\",method =
RequestMethod.GET)

public String placeOrder(@PathVariable String orderId)

{

//模拟service层处理

System.out.println(\"处理订单数据\...\");

//将订单id发送到订单队列

rabbitTemplate.convertAndSend(\"order_exchange\",\"order_routing\",orderId);

return \"下单成功，修改库存\";

}

}