Rabbit MQ 发消息例子以及发失败后定时重发

第一步,创建自己交换器名称,交换器模式等信息

import org.slf4j.Logger;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.util.ErrorHandler;
@Configuration
@EnableRabbit
public class EventConfig {

/**
* 交换器名称
*/
public static final String exchangeName = “aaa.bbb.ccc.ddd”;

@Bean(name = exchangeName)
FanoutExchange exchangeName() {
    return new FanoutExchange(exchangeName);
}

@Primary
@Bean("exchangeName")
RabbitTemplate exchangeNameTemplate(ConnectionFactory connectionFactory) 
### RabbitMQ 消息传递稳定性优化方案 RabbitMQ 是一款功能强大的消息队列中间件,但在实际应用中,为了确保消息传递的稳定性和可靠性,需要综合考虑多种因素。以下是关于 RabbitMQ 消息传递稳定性的优化建议及其具体实施方法。 #### 1. ### 开启 Publisher Confirm 和 Return Mechanism 为了提高消息送的可靠性,可以通过启用 **Publisher Confirm** 或者 **Return Mechanism** 来验证消息是否成功送达目标队列。 - **Publisher Confirm**: 生产者在布消息后会等待 RabbitMQ 的确认通知,只有当 RabbitMQ 成功接收并存储消息时才会返回确认信号[^2]。 - **Return Mechanism**: 如果消息无法路由到任何队列(例如因为绑定键错误),则可以配置 `mandatory` 参数并将未匹配的消息退回给生产者[^2]。 ```java // 配置 publisher confirms channel.confirmSelect(); boolean confirmed = channel.waitForConfirmsOrDie(timeout); if (!confirmed) { throw new IOException("Message not delivered!"); } // 配置 return listener channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> { System.out.println("Unrouted message returned: " + new String(body)); }); ``` #### 2. ### 持久化设置 为了防止 RabbitMQ生宕机或其他意外事件时丢失数据,应将消息和队列都设为持久化状态。这样即使 RabbitMQ 宕机重启,也能恢复之前尚未处理完成的消息[^2]。 ```java // 创建持久化队列 channel.queueDeclare(queueName, true, false, false, null); // 送持久化消息 AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .deliveryMode(2) // 设置 delivery mode 为 2 表示持久化 .build(); channel.basicPublish(exchangeName, routingKey, props, messageBodyBytes); ``` #### 3. ### 手动 Acknowledgment 消费者端应当禁用自动确认机制 (`autoAck=false`),改为显式调用手动确认接口来告知 RabbitMQ 已经成功处理某条消息。这种做法有助于避免因程序崩溃等原因导致部分已消费但未完全处理的消息永久丢失。 ```java // 关闭 auto ack Channel channel = connection.createChannel(); DeliverCallback deliverCallback = (consumerTag, delivery) -> { try { processMessage(delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (Exception e) { channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); } }; channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {}); ``` #### 4. ### 使用惰性队列减少内存占用 对于大规模消息堆积的情况,传统的工作队列可能会消耗过多内存资源从而影响整个系统的性能表现。为此,可以从版本 3.6.0 开始利用懒加载模式(Lazy Queue Mode)创建惰性队列,让大部分消息停留在磁盘而非驻留在 RAM 中,以此达到节省内存的目的[^3]。 ```java @Bean public Queue lazyQueue() { Map<String, Object> args = new HashMap<>(); args.put("x-queue-mode", "lazy"); return QueueBuilder.durable("lazy.queue").withArguments(args).build(); } ``` #### 5. ### 流控与预取参数调整 合理调节消费者的 QoS (Quality of Service)值即 prefetch count 数字大小可以帮助平衡负载均衡以及预防突流量带来的冲击。较小的 prefetched limit 能够促使更多 worker 加入分担任务;反之较大的限额则可加快单一进程内的批量化操作速度[^1]。 ```java int prefetchCount = 10; // 单个消费者最大同时领取的任务数量 channel.basicQos(prefetchCount); ``` --- ###
评论
添加红包

请填写红包祝福语或标题

红包个数最小为10个

红包金额最低5元

当前余额3.43前往充值 >
需支付:10.00
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包
实付
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明:

1.余额是钱包充值的虚拟货币,按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载,可以购买VIP、付费专栏及课程。

余额充值