一、RabbitMQ介绍
RabbitMQ
RabbitMQ是一个开源的消息代理和队列服务器，用来通过普通协议在不同的应用直接共享数据。RabbitMQ使用Erlang语言编写，并且基于AMQP协议实现。
AMQP
AMQP（Advanced Message Queuing Protocol，高级消息队列协议）是一个进程间传递异步消息的网络协议。
发布者（Publisher）发布消息（Message），经由交换机（Exchange）。
交换机根据路由规则将收到的消息分发给与该交换机绑定的队列（Queue）。
最后 AMQP 代理会将消息投递给订阅了此队列的消费者，或者消费者按照需求自行获取。
二、RabbitMQ 各组件功能
1.server：又称broker，接受客户端连接，实现AMQP实体服务。
2.connection：连接和具体broker网络连接。
3.channel：网络信道，几乎所有操作都在channel中进行，channel是消息读写的通道。客户端可以建立多个channel，每个channel表示一个会话任务。
4.message：消息，服务器和应用程序之间传递的数据，由properties和body组成。properties可以对消息进行修饰，比如消息的优先级，延迟等高级特性；body是消息实体内容。
5.Virtual host：虚拟主机，用于逻辑隔离，最上层消息的路由。一个Virtual host可以若干个Exchange和Queue，同一个Virtual host不能有同名的Exchange或Queue。
6.Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列上。
7.banding：Exchange和Queue之间的虚拟连接，binding中可以包括routing key
8.routing key：一个路由规则，虚拟机根据他来确定如何路由 一条消息。
9.Queue：消息队列，用来存放消息的队列。
三、Exchange 类型
3.1 Direct Exchange 直连型交换机
配置队列和交换机持久化以及连接使用设置
/**
* 队列和交换机持久化以及连接使用设置
*/
@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");
}
}
所有发送到Direct Exchange的消息被转发到RouteKey 中指定的Queue。Direct Exchange可以使用默认的默认的Exchange （default Exchange），默认的Exchange会绑定所有的队列，所以Direct可以直接使用Queue名（作为routing key ）绑定。或者消费者和生产者的routing key完全匹配。
3.2 Fanout Exchange 扇型交换机
/**
* 队列和交换机持久化以及连接使用设置
*/
@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());
}
}
不处理路由键，只需简单的将队列绑定到交换机上。发送到改交换机上的消息都会被发送到与该交换机绑定的队列上。Fanout转发是最快的。
3.3 Topic Exchange 主题交换机
@Configuration
public class TopicRabbitConfig {
//绑定键
public final static String apple = "topic.apple";
public final static String pear = "topic.pear";
public final static String fruit = "topic.fruit";
@Bean
public Queue firstQueue() {
return new Queue(TopicRabbitConfig.apple);
}
@Bean
public Queue secondQueue() {
return new Queue(TopicRabbitConfig.pear);
}
@Bean
public Queue thirdQueue() {
return new Queue(TopicRabbitConfig.fruit);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
//这样只要是消息携带的路由键是topic.apple,才会分发到该队列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstQueue()).to(exchange()).with(apple);
}
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondQueue()).to(exchange()).with(pear);
}
//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
@Bean
Binding bindingExchangeMessage3() {
return BindingBuilder.bind(thirdQueue()).to(exchange()).with("topic.#");
}
}
发送到Topic Exchange的消息被转发到所有关心的Routing key中指定topic的Queue上。Exchange 将routing key和某Topic进行模糊匹配，此时队列需要绑定一个topic。所谓模糊匹配就是可以使用通配符，“#”可以匹配一个或多个词，“”只匹配一个词比如“log.#”可以匹配“log.info.test” "log. "就只能匹配log.error。
四、死信队列 DLX
4.1 死信队列介绍
“死信”是RabbitMQ中的一种消息机制，配置了死信队列信息，那么该消息将会被丢进死信队列中。否则该消息将会被丢弃。
当你在消费消息时，如果队列里的消息出现以下情况消息将成为“死信”：
1.消息被拒绝，使用 basicNack / basicReject ，并且此时 requeue = false
2.消息TTL过期（超时时间）
3.消息队列的消息数量已经超过最大队列长度
4.2 死信队列配置
Step1.配置业务队列，绑定到业务交换机上
Step2.为业务队列配置死信交换机和路由key
Step3.为死信交换机配置死信队列
每个使用死信的业务队列都需要配置一个死信交换机，这里同一个项目的死信交换机可以共用一个，然后为每个业务队列分配一个单独的路由key。
@Configuration
public class RabbitmqConfig {
/**
* 定义死信队列相关信息
*/
public final static String DEAD_QUEUE = "dead_queue";
/**
* 死信路由键
*/
public final static String DEAD_ROUTING_KEY = "dead_routing_key";
/**
* 死信交换机
*/
public final static String DEAD_EXCHANGE = "dead_exchange";
/**
* 死信队列 交换机标识符
*/
public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
/**
* 死信队列交换机绑定键标识符
*/
public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
/**
* 邮件队列
*/
private static final String FANOUT_EMAIL_QUEUE = "fanout_email_queue";
/**
* fanout 交换机
*/
private static final String EXCHANGE_NAME = "fanoutExchange";
/**
* 定义邮件队列
* 注意：如若队列已被创建，是无法再次修改，这时绑定死信队列会报错
*/
@Bean
public Queue fanOutEamilQueue() {
// 将普通队列绑定到死信队列交换机上
Map<String, Object> args = new HashMap<>(2);
//x-dead-letter-exchange 声明当前业务队列需要绑定的死信交换机
args.put(DEAD_LETTER_QUEUE_KEY, DEAD_EXCHANGE);
//x-dead-letter-routing-key 声明当前信息转到死信队列后的路由键
args.put(DEAD_LETTER_ROUTING_KEY, DEAD_ROUTING_KEY);
return QueueBuilder.durable(FANOUT_EMAIL_QUEUE).withArguments(args).build();
}
/**
* 定义交换机
*/
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange(EXCHANGE_NAME);
}
/**
* 邮件队列与交换机绑定
* Param1：fanoutEmailQueue作为bean的ID，查询注册到Spring容器中对应ID的bean
* 参数名称必须与队列和交换机名称相同
*/
@Bean
Binding bindingExchangeEamil(Queue fanOutEamilQueue, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanOutEamilQueue).to(fanoutExchange);
}
/**
* 创建配置死信邮件队列
*/
@Bean
public Queue deadQueue() {
Queue queue = new Queue(DEAD_QUEUE, true);
return queue;
}
/**
* 创建死信交换机
*/
@Bean
public DirectExchange deadExchange() {
return new DirectExchange(DEAD_EXCHANGE);
}
/**
* 死信队列与死信交换机绑定
*/
@Bean
public Binding bindingDeadExchange(Queue deadQueue, DirectExchange deadExchange) {
return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY);
}
}
4.3 应用场景
一般用在较为重要的业务队列中，确保未被正确消费的消息不被丢弃，一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常，处理过程中参数校验异常，或者因网络波动导致的查询异常等等，当发生异常时，当然不能每次通过日志来获取原消息，然后让运维帮忙重新投递消息。通过配置死信队列，可以让未正确处理的消息暂存到另一个队列中，待后续排查清楚问题后，编写相应的处理代码来处理死信消息，这样比手工恢复数据要好太多了。
五、消费端 ACK、NACK、REJECT 与 重回队列
channel.basicAck(deliveryTag, false); 成功
channel.basicReject(deliveryTag, false); 失败
channel.basicNack(deliveryTag, true, false); 失败
消费端进行消费的时候，如果由于业务异常我们可以进行日志的记录，然后进行补偿！（也可以加上最大努力次数的尝试）如果由于服务器宕机等严重问题，那我们就需要手动进行ack保证消费端的消费成功
/**
* deliveryTag：消息在mq中的唯一标识
* multiple：是否批量
* requeue：是否需要重回队列
*/
public void basicNack(long deliveryTag,boolean multiple, boolean requeue);
basicReject 与 basicNack 的区别是：basicReject 一次只能拒绝单条消息
重回队列就是为了对没有处理成功的消息，把消息重新投递给broker。实际应用中一般都不开启重回队列。
六、生产者 Confirm 确认消息
消息的确认，指生产者收到投递消息后，如果Broker收到消息就会给我们 的生产者一个应答，生产者接受应答来确认broker是否收到消息。
AMQP 事务机制也可以用来保证消息的准确到达，但在一定程度上是消耗了性能的，所以不推荐使用。
与 AMQP 事务不同的是 Confirm 是针对一条消息的，而事务是可以针对多条消息的。
Confirm 模式最大的好处在于它是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息。
@Component
public class RabbitmqProducer implements RabbitTemplate.ConfirmCallback{
/**
* 确认后回调
* producer->rabbitmq broker cluster 则会返回一个 confirmCallback
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
System.out.println("---------Send ack Success---------");
}else {
System.out.println("---------Send ack fail---------");
}
if (!StringUtils.isEmpty(cause)){
System.out.println("失败原因："+cause);
}
}
}
七、生产者 Return 返回消息
Return消息机制处理一些不可路由的消息，我们的生产者通过指定一个Exchange和Routinkey，把消息送达到某一个队列中去，然后我们消费者监听队列进行消费处理！
在某些情况下，如果我们在发送消息的时候当Exchange不存在或者指定的路由key路由找不到，这个时候如果我们需要监听这种不可到达的消息，就要使用Return Listener！
Mandatory=true则会监听器会接受到路由不可达的消息，然后处理。如果设置为false，broker将会自动删除该消息。
@Component
public class RabbitmqProducer implements RabbitTemplate.ReturnCallback{
/**
* 失败后return回调
* exchange->queue 投递失败则会返回一个 returnCallback
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
System.out.println("---------Send fail---------");
System.out.println("消息主体："+message);
System.out.println("回应码："+replyCode);
System.out.println("回应描述："+replyText);
System.out.println("消息使用的交换器 exchange："+exchange);
System.out.println("消息使用的路由键 routing："+routingKey);
}
}
八、消费端限流
假设我们有个场景，首先，我们有个rabbitMQ服务器上有上万条消息未消费，然后我们随便打开一个消费者客户端，会出现：巨量的消息瞬间推送过来，但是我们的消费端无法同时处理这么多数据。
这时就会导致你的服务崩溃。其他情况也会出现问题，比如你的生产者与消费者能力不匹配，在高并发的情况下生产端产生大量消息，消费端无法消费那么多消息。
rabbitMQ提供了一种qos（服务质量保证）的功能，即非自动确认消息的前提下，如果有一定数目的消息（通过consumer或者Channel设置qos）未被确认，不进行新的消费。
void basicQOS(unit prefetchSize,ushort prefetchCount,Boolean global)方法。
prefetchSize:0 单条消息的大小限制。0就是不限制，一般都是不限制。
prefetchCount: 设置一个固定的值，告诉rabbitMQ不要同时给一个消费者推送多余N个消息，即一旦有N个消息还没有ack，则consumer将block掉，直到有消息ack
global：truefalse 是否将上面的设置用于channel，也是就是说上面设置的限制是用于channel级别的还是consumer的级别的。
九、消费端幂等性
消息幂等性，其实就是保证同一个消息不被消费者重复消费两次。当消费者消费完消息之后，通常会发送一个ack应答确认信息给生产者，但是这中间有可能因为网络中断等原因，导致生产者未能收到确认消息，由此这条消息将会被 重复发送给其他消费者进行消费，实际上这条消息已经被消费过了，这就是重复消费的问题。
如何避免重复消费的问题？
消息全局ID或者写个唯一标识。每次消费消息之前根据消息id去数据库中查看是否存在，如果存在则数据已消费，不再进行处理。否则正常消费消息，并且进行入库操作。(消息全局ID作为数据库表的主键，防止重复)
利用Redis的setnx 命令：给消息分配一个全局ID，只要消费过该消息，将 < id,message>以K-V键值对形式写入redis，消费者开始消费前，先去redis中查询有没消费记录即可。
示例1：
public void checkMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
//获取MessageID
String messageId = message.getMessageProperties().getMessageId();
//查数据库中是否消费了当前消息
Message message = messageService.getMessageId();
//当前信息未被消费
if (ObjectUtils.isEmpty(message)) {
//获取消费内容
String msg = new String(message.getBody(), StandardCharsets.UTF_8);
//手动ACK
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//存入到表中，标识该消息已消费
messageService.addMessage(new Message() {{
setMessageId(messageId);
}});
} else {
System.out.println("该消息已消费");
}
}
十、生产端的消息可靠性
消息落库
消息延迟投递
在高并发场景下，每次进行db的操作都是非常消耗性能的。我们使用延迟队列来减少一次数据库的操作。
1.发送者发送信息到MQ，消费者为下游业务方。
1.1成功后，发送者发送信息值MQ，消费者为回调服务方。
1.1.1回调服务接受数据后，落库
1.2 失败，等待发送者的延时投递消息。
2.发送者发送延迟投递消息到MQ，消费者为回调服务。
2.1 查库，确认下游服务端消费已成功
2.2 查库，确认下游服务消费失败，通过RPC调用发送者的接口重新发送
十一、延迟投递
场景一：淘宝七天自动确认收货。在我们签收商品后，物流系统会在七天后延时发送一个消息给支付系统，通知支付系统将款打给商家，这个过程持续七天，就是使用了消息中间件的延迟推送功能。
场景二：2306 购票支付确认页面。我们在选好票点击确定跳转的页面中往往都会有倒计时，代表着 30 分钟内订单不确认的话将会自动取消订单。其实在下订单那一刻开始购票业务系统就会发送一个延时消息给订单系统，延时30分钟，告诉订单系统订单未完成，如果我们在30分钟内完成了订单，则可以通过逻辑代码判断来忽略掉收到的消息。
上述类似的需求是我们经常会遇见的问题。最常用的方法是定期轮训数据库，设置状态。在数据量小的时候并没有什么大的问题，但是数据量一大轮训数据库的方式就会变得特别耗资源。当面对千万级、上亿级数据量时，本身写入的IO就比较高，导致长时间查询或者根本就查不出来，更别说分库分表以后了。除此之外，还有优先级队列，基于优先级队列的JDK延迟队列，时间轮等方式。但如果系统的架构中本身就有RabbitMQ的话，那么选择RabbitMQ来实现类似的功能也是一种选择。
11.1 Delayed Message 插件实现
Error：reply-code=503, reply-text=COMMAND_INVALID - unknown exchange type 'x-delayed-message'.
原因：没有找到对应 x-delayed-message 的 exchange type
下载rabbitMQ插件 rabbitmq_delayed_message_exchange，根据Rabbitmq版本选择对应插件版本，并放入Rabbitmq-Plugins。
Rabbitmq 3.7x 3.8x 通用插件：链接：https://pan.baidu.com/s/1fZImyyGWb20NkDgLwPsNYA 提取码：km1k
启用插件后重启mq
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
配置 x-delayed-message Exchange
@Configuration
public class DelayConfig {
@Bean
public CustomExchange delayExchange() {
Map<String, Object> args = new HashMap<>();
//设置 Exchange 的 Arguments ， 交换机类型可以为 direct ,topic ,fanout
args.put("x-delayed-type", "direct");
return new CustomExchange("delayExchange", "x-delayed-message",true, false, args);
}
@Bean
public Queue delayQueue() {
return new Queue("delayQueue", true);
}
@Bean
public Binding lazyBinding() {
return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("delay_msg").noargs();
}
}
发送消息并设置延时时间
rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//设置消息持久化
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
//设置延迟发送时间 10s
message.getMessageProperties().setHeader("x-delay", "10000");
return message;
}
}, correlationId);
11.2 死信队列 + TTL 实现
一个消息比在同一队列中的其他消息提前过期，提前过期的也不会优先进入死信队列，它们还是按照入库的顺序让消费者消费。如果第一进去的消息过期时间是1小时，那么死信队列的消费者也许等1小时才能收到第一个消息。
所以在考虑使用RabbitMQ来实现延迟任务队列的时候，需要确保业务上每个任务的延迟时间是一致的。如果遇到不同的任务类型需要不同的延时的话，需要为每一种不同延迟时间的消息建立单独的消息队列。
十二、消费端自动重试
创建测试队列，并绑定至死信队列
/**
* 队列和交换机持久化以及连接使用设置
*/
@Configuration
public class MqConfig {
/**
* 定义死信队列相关信息
*/
public final static String DEAD_QUEUE = "dead_queue";
/**
* 死信路由键
*/
public final static String DEAD_ROUTING_KEY = "dead_routing_key";
/**
* 死信交换机
*/
public final static String DEAD_EXCHANGE = "dead_exchange";
/**
* 死信队列 交换机标识符
*/
public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
/**
* 死信队列交换机绑定键标识符
*/
public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
/**
* 邮件队列
*/
private static final String TEST_QUEUE = "test_queue";
/**
* fanout 交换机
*/
private static final String TEST_EXCHANGE = "test_exchange";
/**
* 创建业务队列，并将业务队列绑定到死信队列交换机上
* 注意：如若队列已被创建，是无法再次修改，这时绑定死信队列会报错
*/
@Bean
public Queue testQueue() {
// 将普通队列绑定到死信队列交换机上
Map<String, Object> args = new HashMap<>(2);
//x-dead-letter-exchange --> 绑定到指定的 死信交换机
args.put(DEAD_LETTER_QUEUE_KEY, DEAD_EXCHANGE);
//x-dead-letter-routing-key --> 信息转到死信队列后的路由键
args.put(DEAD_LETTER_ROUTING_KEY, DEAD_ROUTING_KEY);
// Queue queue = new Queue(FANOUT_EMAIL_QUEUE, true, false, false, args); return queue;
//为业务队列配置参数
return QueueBuilder.durable(TEST_QUEUE).withArguments(args).build();
}
/**
* 创建业务交换机
*/
@Bean
FanoutExchange testExchange() {
return new FanoutExchange(TEST_EXCHANGE);
}
/**
* 业务队列与交换机绑定
* Param1：fanoutEmailQueue作为bean的ID，查询注册到Spring容器中对应ID的bean
* 参数名称必须与队列和交换机名称相同
*/
@Bean
Binding bindingTestExchange(Queue testQueue, FanoutExchange testExchange) {
return BindingBuilder.bind(testQueue).to(testExchange);
}
/**
* 创建死信队列
*/
@Bean
public Queue deadQueue() {
return new Queue(DEAD_QUEUE, true);
}
/**
* 创建死信交换机
*/
@Bean
public DirectExchange deadExchange() {
return new DirectExchange(DEAD_EXCHANGE);
}
/**
* 死信队列与死信交换机绑定
*/
@Bean
public Binding bindingDeadExchange(Queue deadQueue, DirectExchange deadExchange) {
return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY);
}
}
消费者拦截器的容器工厂
/**
* 实现消费者拦截器的容器工厂，增加 重试类 来设置重试参数
*
* 本例子，获取了Rabbitproperties 的配置，也可以自定义。 消息在被消费者获取后，重试3次后，将会过期，
* 我设置了死信队列，将重试处理不了的消息 了 路由到了死信队列中 。死信队列的消息需要人工干预处理。
*/
@Slf4j
@Configuration
public class RabbitRetryConfig {
@Autowired
ConnectionFactory rabbitConnectionFactory;
//@Bean 缓存连接池
//public CachingConnectionFactory rabbitConnectionFactory
@Autowired
RabbitProperties properties;
// 存在此名字的bean 自带的容器工厂会不加载，如果想自定义来区分开 需要改变bean 的名称
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
containerFactory.setConnectionFactory(rabbitConnectionFactory);
// 并发消费者数量
containerFactory.setConcurrentConsumers(1);
containerFactory.setMaxConcurrentConsumers(20);
// 自动应答
containerFactory.setAcknowledgeMode(AcknowledgeMode.AUTO);
//containerFactory.setMessageConverter(new Jackson2JsonMessageConverter());
containerFactory.setChannelTransacted(true);
containerFactory.setAdviceChain(
RetryInterceptorBuilder
.stateless()
.recoverer(new RejectAndDontRequeueRecoverer())
.retryOperations(rabbitRetryTemplate())
.build()
);
return containerFactory;
}
@Bean
public RetryTemplate rabbitRetryTemplate() {
RetryTemplate retryTemplate = new RetryTemplate();
// 设置监听（不是必须）
retryTemplate.registerListener(new RetryListener() {
@Override
public <T, E extends Throwable> boolean open(RetryContext retryContext, RetryCallback<T, E> retryCallback) {
// 执行之前调用 （返回false时会终止执行）
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
// 重试结束的时候调用 （最后一次重试 ）
}
@Override
public <T, E extends Throwable> void onError(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
// 异常 都会调用
log.error("-----第{}次调用", retryContext.getRetryCount());
}
});
// 个性化处理异常和重试 （不是必须）
/* Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<>();
//设置重试异常和是否重试
retryableExceptions.put(AmqpException.class, true);
//设置重试次数和要重试的异常
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(5,retryableExceptions);*/
retryTemplate.setBackOffPolicy(backOffPolicyByProperties());
retryTemplate.setRetryPolicy(retryPolicyByProperties());
return retryTemplate;
}
@Bean
public ExponentialBackOffPolicy backOffPolicyByProperties() {
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
long maxInterval = properties.getListener().getSimple().getRetry().getMaxInterval().getSeconds();
long initialInterval = properties.getListener().getSimple().getRetry().getInitialInterval().getSeconds();
double multiplier = properties.getListener().getSimple().getRetry().getMultiplier();
// 重试间隔
backOffPolicy.setInitialInterval(initialInterval * 1000);
// 重试最大间隔
backOffPolicy.setMaxInterval(maxInterval * 1000);
// 重试间隔乘法策略
backOffPolicy.setMultiplier(multiplier);
return backOffPolicy;
}
@Bean
public SimpleRetryPolicy retryPolicyByProperties() {
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
int maxAttempts = properties.getListener().getSimple().getRetry().getMaxAttempts();
retryPolicy.setMaxAttempts(maxAttempts);
return retryPolicy;
}
}
rabbitmq:
listener:
simple:
retry:
enabled: true #开启消费者重试 消费者发生异常会进行重试
max-attempts: 3 #最大重试次数
initial-interval: 1000 #重试间隔时间
十三、RabbitMQ 安装
Erlang安装
Rabbitmq是基于erlang语言开发的，所以必须先安装erlang。
链接：https://pan.baidu.com/s/1BxbedYTAHOkUDinCp90Ung 提取码：fr8m
tar -zxvf otp_src_22.0.tar.gz
cd /usr/local/otp_src_22.0/
# erlang编译安装默认是装在/usr/local下的bin和lib中，我们将他统一装到/usr/local/erlang中，方便查找和使用。
mkdir ../erlang
#配置安装路径
./configure --prefix=/usr/local/erlang
APPLICATIONS INFORMATION 与 DOCUMENTATION INFORMATION是正常的，不影响正常编译。
编译安装
make && make install
安装后，/usr/local/erlang中就会出现如下：
添加到环境变量，并重载
PATH=$PATH:/usr/local/erlang/bin
测试输入erl ; 输入halt(). 退出
RabbltMQ安装
链接：https://pan.baidu.com/s/1iL-Dl5oFQngSFu0HUCWncQ 提取码：p402
由于是tar.xz格式的所以需要用到xz，没有的话就先安装
yum install -y xz
第一次解压
/bin/xz -d rabbitmq-server-generic-unix-3.7.15.tar.xz
第二次解压
tar -xvf rabbitmq-server-generic-unix-3.7.15.tar
配置环境变量后,刷新环境变量
export PATH=$PATH:/usr/local/rabbitmq/sbin
创建配置目录
mkdir /etc/rabbitmq
添加web管理插件
rabbitmq-plugins enable rabbitmq_management
启动命令
#启动
rabbitmq-server -detached
#停止
rabbitmqctl stop
#查看状态
rabbitmqctl status
用户配置
查看所有用户
rabbitmqctl list_users
添加一个用户
rabbitmqctl add_user admin rabbit
配置权限
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
查看用户权限
rabbitmqctl list_user_permissions admin
设置tag
rabbitmqctl set_user_tags admin administrator
删除用户（安全起见，删除默认用户）
rabbitmqctl delete_user guest
设置后重启rabbitmq，并用admin用户登录
查看控制台
访问：http://127.0.0.1:15672/
账号： admin rabbit
