springboot整合rabbitmq，开启手工确认。保证消息100%投递

package com.example.rabbitdemo.config; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.AcknowledgeMode; 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.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import com.example.rabbitdemo.consumer.RabbitReceiver; @Configuration public class RabbitConfig { private static Logger logger = LoggerFactory.getLogger(RabbitConfig.class); @Autowired private CachingConnectionFactory connectionFactory; @Autowired private RabbitReceiver rabbitReceiver; /** * 创建队列 */ @Bean public Queue queueForSingleMode() { return new Queue("quene.acknowledgemode.manualtest"); } /** * 创建主题交换机 */ @Bean public TopicExchange createTopicExchange() { return new TopicExchange("opic.exchange.test"); } /** * 将队列与交换机进行绑定 */ @Bean public Binding binding() { return BindingBuilder.bind(queueForSingleMode()).to(createTopicExchange()).with("manualtest"); } @Bean public RabbitTemplate rabbitTemplate() { RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter()); /** * producer ---> exchange 消息发送确认回调方法 * 如果消息没有到exchange,则confirm回调,ack=false * 如果消息到达exchange,则confirm回调,ack=true */ rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { StringBuffer sBuffer = new StringBuffer(); sBuffer.append("correlationId = ").append(correlationData.getId()).append(" ,是否消息发送确认成功 = ").append(ack) .append(""); if (!ack) { sBuffer.append("消息失败原因: ").append(cause); } logger.info("producer ---> exchange 消息confirm回调: {}", sBuffer.toString()); }); // 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调 rabbitTemplate.setMandatory(true); /** * exchange --> queue 消息发送确认回调方法 * exchange到queue成功,则不回调return * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了) */ rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { StringBuffer sBuffer = new StringBuffer(); sBuffer.append("return--message:").append(new String(message.getBody())).append(",replyCode:") .append(replyCode).append(",replyText:").append(replyText).append(",exchange:").append(exchange) .append(",routingKey:").append(",routingKey:").append(routingKey); logger.info("exchange --> queue 消息return回调: {}", sBuffer.toString()); }); return rabbitTemplate; } /** * AcknowledgeMode.AUTO ，它会根据方法的执行情况来决定是否确认还是拒绝（是否重新入queue） * 如果消息成功被消费（成功的意思是在消费的过程中没有抛出异常），则自动确认 * 当抛出 AmqpRejectAndDontRequeueException 异常的时候，则消息会被拒绝，且 requeue = false（不重新入队列） * 当抛出 ImmediateAcknowledgeAmqpException 异常，则消费者会被确认 * 其他的异常，则消息会被拒绝，且 requeue = true（如果此时只有一个消费者监听该队列， * 则有发生死循环的风险，多消费端也会造成资源的极大浪费，这个在开发过程中一定要避免的）。 * 可以通过 setDefaultRequeueRejected（默认是true）去设置 * @param connectionFactory * @return */ @Bean public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setQueues(queueForSingleMode());// 监听的队列 container.setExposeListenerChannel(true); container.setMaxConcurrentConsumers(30);//每个listener最大并发消费者数,对那种对消息的顺序有苛刻要求的场景不适合并发消费 container.setConcurrentConsumers(1);//对每个listener在初始化的时候设置的并发消费者数量 /** * prefetch是每次从一次性从broker里面取的待消费的消息的个数<br> * 在使用了acknowledge消息确认机制情况下，使用prefetch_count进行流量控制 */ container.setPrefetchCount(30); container.setConnectionFactory(connectionFactory); container.setMessageConverter(new Jackson2JsonMessageConverter()); container.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 程序手动确认 container.setMessageListener(rabbitReceiver); return container; } /** * 使用 JSON 序列化化消息 * @param connectionFactory * @return */ @Bean public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(new Jackson2JsonMessageConverter()); /** *开启手动确认处理消息 */ factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); return factory; } }