掌控消息全链路(4)——RabbitMQ/Spring-AMQP高级特性详解之事务与消息分发

🔥我的主页：九转苍翎⭐️个人专栏：《Java SE》《Java集合框架系统精讲》《MySQL高手之路：从基础到高阶》《计算机网络》《Java工程师核心能力体系构建》《RabbitMQ理论与实践》天行健，君子以自强不息。

1.事务

AMQP(高级消息队列协议)实现了事务机制，主要用于确保消息的原子性发布和确认。换言之，它允许你将多个操作(如发送消息、确认消息)绑定在一起，要么全部成功，要么全部失败

发送消息

@RestController@RequestMapping("/producer")publicclassProducerController{@Resource(name ="transRabbitTemplate")privateRabbitTemplate transRabbitTemplate;@Transactional@RequestMapping("/trans")publicStringtrans(){ transRabbitTemplate.convertAndSend("",Constants.TRANS_QUEUE,"trans test ---> 1");int num =5/0; transRabbitTemplate.convertAndSend("",Constants.TRANS_QUEUE,"trans test ---> 2");return"发送成功";}}

Spring Boot的RabbitMQ自动配置默认会确认模式，但RabbitMQ不允许同一个通道同时使用事务模式和确认模式，所以需要确保publisher confirms被禁用

spring:rabbitmq:publisher-confirm-type: none publisher-returns:false

配置RabbitTemplate和事务管理器

@ConfigurationpublicclassRabbitTemplateConfig{@Bean("transRabbitTemplate")publicRabbitTemplatetransRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate =newRabbitTemplate(connectionFactory); rabbitTemplate.setChannelTransacted(true);return rabbitTemplate;}@BeanpublicRabbitTransactionManagerrabbitTransactionManager(ConnectionFactory connectionFactory){returnnewRabbitTransactionManager(connectionFactory);}}

配置队列

@ConfigurationpublicclassRabbitMQConfig{@Bean("transQueue")publicQueuetransQueue(){returnQueueBuilder.durable(Constants.TRANS_QUEUE).build();}}

2.消息限流

消息限流(Flow Control) 是RabbitMQ防止生产者发送消息速度超过消费者处理能力，导致消息积压和系统崩溃的保护机制

发送消息

@RestController@RequestMapping("/producer")publicclassProducerController{@Resource(name ="rabbitTemplate")privateRabbitTemplate rabbitTemplate;@RequestMapping("/qos")publicStringqos(){for(int i =0; i <20; i++){ rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE,"qos","qos test:"+ i);}return"发送成功";}}

配置消费者

@Component@Slf4jpublicclassQosListener{@RabbitListener(queues =Constants.QOS_QUEUE)publicvoidhandMessage(Message message,Channel channel)throwsIOException{long deliveryTag = message.getMessageProperties().getDeliveryTag();try{ log.info("接收到消息:{},deliveryTag:{}",newString(message.getBody(),StandardCharsets.UTF_8), deliveryTag); log.info("处理成功");// channel.basicAck(deliveryTag, true); // 消费者不确认消息}catch(Exception e){ channel.basicNack(deliveryTag,true,true);}}}

限制每个消费者未确认的最大消息数

spring:rabbitmq:listener:simple:acknowledge-mode: manual # 消费者确认机制prefetch:5

声明和配置交换器、队列和绑定关系

@ConfigurationpublicclassRabbitMQConfig{@Bean("qosQueue")publicQueueqosQueue(){returnQueueBuilder.durable(Constants.QOS_QUEUE).build();}@Bean("qosExchange")publicDirectExchangeqosExchange(){returnExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();}@Bean("qosBinding")publicBindingqosBinding(@Qualifier("qosExchange")DirectExchange directExchange,@Qualifier("qosQueue")Queue queue){returnBindingBuilder.bind(queue).to(directExchange).with("qos");}}