RabbitMQ高级

淩亂°似流年 2023-09-25 19:51 146阅读 0赞

针对消息可能丢失的问题，RabbitMQ分别给出了解决方案：  
\- 生产者确认机制  
RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功  
\- publisher-confirm，发送者确认  
\- 消息成功投递到交换机，返回ack  
\- 消息未投递到交换机，返回nack  
\- publisher-return，发送者回执  
\- 消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。  
设置回调  
@Slf4j  
@Configuration  
public class CommonConfig implements ApplicationContextAware \{  
@Override  
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException \{  
// 获取RabbitTemplate  
RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);  
// 设置ReturnCallback  
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> \{  
// 投递失败，记录日志  
log.info("消息发送失败，应答码\{\}，原因\{\}，交换机\{\}，路由键\{\},消息\{\}",  
replyCode, replyText, exchange, routingKey, message.toString());  
// 如果有业务需要，可以重发消息  
\});  
\}  
\}

\- mq持久化  
@Bean  
public DirectExchange simpleExchange()\{  
// 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除  
return new DirectExchange("simple.direct", true, false);  
\}  
默认情况下，由SpringAMQP声明的交换机都是持久化的

\- 消费者确认机制  
RabbitMQ是阅后即焚机制，RabbitMQ确认消息被消费者消费后会立刻删除。  
而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理消息。  
而SpringAMQP则允许配置三种确认模式：  
•manual：手动ack，需要在业务代码结束后，调用api发送ack。  
•auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack 默认模式  
•none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

\- 失败重试机制  
spring:  
rabbitmq:  
listener:  
simple:  
retry:  
enabled: true \# 开启消费者失败重试  
initial-interval: 1000 \# 初识的失败等待时长为1秒  
multiplier: 1 \# 失败的等待时长倍数，下次等待时长 = multiplier \* last-interval  
max-attempts: 3 \# 最大重试次数  
stateless: true \# true无状态；false有状态。如果业务中包含事务，这里改为false

失败策略：定义一个队列接收消费失败的消息  
@Bean  
public DirectExchange errorMessageExchange()\{  
return new DirectExchange("error.direct");  
\}  
@Bean  
public Queue errorQueue()\{  
return new Queue("error.queue", true);  
\}  
@Bean  
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange)\{  
return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");  
\}

@Bean  
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate)\{  
return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");  
\}

死信交换机

TTL  
一个队列中的消息如果超时未消费，则会变为死信，超时分为两种情况：  
\- 消息所在的队列设置了超时时间  
\- 消息本身设置了超时时间

TTL加死信交换机 定时任务  
@Bean  
public Queue ttlQueue()\{  
return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化  
.ttl(10000) // 设置队列的超时时间，10秒  
.deadLetterExchange("dl.ttl.direct") // 指定死信交换机  
.build();  
\}

普通集群：是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。  
•镜像集群：是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性