RabbitMQ可靠投递之confirmCallback确认模式------RabbitMQ

1. 保证消息不丢失,使用事务消息,会导致性能下降250倍,为此需要使用确认机制
2. confirmCallback确认模式,从消息生产者到服务器交换机
3. returnCallback若交换机未能投递到队列,需要重新回退到交换机内
4. 开启发送端确认后
5. 如果没有投递成功,消息会被记录和保存的

复制代码

保证消息不丢失,利用事件消息,会导致性能下降250倍,为此需要利用确认机制
confirmCallback确认模式,从消息生产者到服务器交换机
returnCallback若交换机未能投递到队列,需要重新回退到交换机内
开开导送端确认后
如果没有投递成功,消息会被记载和保存的
   

1. spring.application.name=Mall-order
2. spring.cloud.nacos.config.server-addr=127.0.0.1:8848
3. spring.cloud.nacos.discovery.server-addr=127.0.0.1:8848
4. spring.rabbitmq.host=192.168.56.10
5. spring.rabbitmq.port=5672
6. spring.rabbitmq.virtual-host=/
7. #只要抵达队列,以异步形式回调
8. spring.rabbitmq.template.mandatory=true
9. #新版本的发送确认
10. spring.rabbitmq.publisher-confirm-type=CORRELATED
11. spring.cloud.nacos.config.namespace=becc5563-6c9b-47ff-a6b1-580982722371

复制代码

1. spring.application.name=Mall-order
2. spring.cloud.nacos.config.server-addr=127.0.0.1:8848
3. spring.cloud.nacos.discovery.server-addr=127.0.0.1:8848
4. spring.rabbitmq.host=192.168.56.10
5. spring.rabbitmq.port=5672
6. spring.rabbitmq.virtual-host=/
7. #只要抵达队列,以异步形式回调
8. spring.rabbitmq.template.mandatory=true
9. #新版本的发送确认
10. spring.rabbitmq.publisher-confirm-type=CORRELATED
11. spring.cloud.nacos.config.namespace=becc5563-6c9b-47ff-a6b1-580982722371

复制代码

  

1. package com.alatus.mall.order.controller;
3. import com.alatus.mall.order.entity.OrderEntity;
4. import com.alatus.mall.order.entity.OrderReturnReasonEntity;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.beans.factory.annotation.Autowired;
7. import org.springframework.web.bind.annotation.GetMapping;
8. import org.springframework.web.bind.annotation.RestController;
10. import java.util.Date;
11. import java.util.UUID;
13. @RestController
14. public class RabbitController {
15.     @Autowired
16.     private RabbitTemplate rabbitTemplate;
17.     @GetMapping("/sendMessage")
18.     public String sendMessage(){
19.         for (int i = 0; i < 10; i++) {
20.             if(i % 2 == 0){
21.                 OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
22.                 reasonEntity.setId(1L);
23.                 reasonEntity.setCreateTime(new Date());
24.                 reasonEntity.setName("哈哈"+i);
25.                 rabbitTemplate.convertAndSend("helloWorld","hello",reasonEntity);
26.             }
27.             else{
28.                 OrderEntity entity = new OrderEntity();
29.                 entity.setOrderSn(UUID.randomUUID().toString());
30.                 rabbitTemplate.convertAndSend("helloWorld","hello",entity);
31.             }
32.         }
33.         return "OK";
34.     }
35. }

复制代码

1. package com.alatus.mall.order.controller;
3. import com.alatus.mall.order.entity.OrderEntity;
4. import com.alatus.mall.order.entity.OrderReturnReasonEntity;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.beans.factory.annotation.Autowired;
7. import org.springframework.web.bind.annotation.GetMapping;
8. import org.springframework.web.bind.annotation.RestController;
10. import java.util.Date;
11. import java.util.UUID;
13. @RestController
14. public class RabbitController {
15.     @Autowired
16.     private RabbitTemplate rabbitTemplate;
17.     @GetMapping("/sendMessage")
18.     public String sendMessage(){
19.         for (int i = 0; i < 10; i++) {
20.             if(i % 2 == 0){
21.                 OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
22.                 reasonEntity.setId(1L);
23.                 reasonEntity.setCreateTime(new Date());
24.                 reasonEntity.setName("哈哈"+i);
25.                 rabbitTemplate.convertAndSend("helloWorld","hello",reasonEntity);
26.             }
27.             else{
28.                 OrderEntity entity = new OrderEntity();
29.                 entity.setOrderSn(UUID.randomUUID().toString());
30.                 rabbitTemplate.convertAndSend("helloWorld","hello",entity);
31.             }
32.         }
33.         return "OK";
34.     }
35. }

复制代码

  

1. package com.alatus.mall.order.config;
3. import org.springframework.amqp.rabbit.connection.CorrelationData;
4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
5. import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
6. import org.springframework.amqp.support.converter.MessageConverter;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.context.annotation.Bean;
9. import org.springframework.context.annotation.Configuration;
11. import javax.annotation.PostConstruct;
13. @Configuration
14. public class RabbitConfig {
15.     @Autowired
16.     private RabbitTemplate rabbitTemplate;
17.     @Bean
18.     public MessageConverter messageConverter(){
19.         return new Jackson2JsonMessageConverter();
20.     }
21. //    定制RabbitCallback,构造器创建完成,执行此方法
22.     @PostConstruct
23.     public void initRabbitTemplate(){
24. //        设置确认回调
25.         rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
26.             @Override
27.             public void confirm(CorrelationData correlationData, boolean ack, String cause) {
28.                 System.out.println(""+correlationData+ack+cause);
29.             }
30.         });
31.     }
32. }

复制代码

1. package com.alatus.mall.order.config;
3. import org.springframework.amqp.rabbit.connection.CorrelationData;
4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
5. import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
6. import org.springframework.amqp.support.converter.MessageConverter;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.context.annotation.Bean;
9. import org.springframework.context.annotation.Configuration;
11. import javax.annotation.PostConstruct;
13. @Configuration
14. public class RabbitConfig {
15.     @Autowired
16.     private RabbitTemplate rabbitTemplate;
17.     @Bean
18.     public MessageConverter messageConverter(){
19.         return new Jackson2JsonMessageConverter();
20.     }
21. //    定制RabbitCallback,构造器创建完成,执行此方法
22.     @PostConstruct
23.     public void initRabbitTemplate(){
24. //        设置确认回调
25.         rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
26.             @Override
27.             public void confirm(CorrelationData correlationData, boolean ack, String cause) {
28.                 System.out.println(""+correlationData+ack+cause);
29.             }
30.         });
31.     }
32. }

复制代码

  

1. package com.alatus.mall.order.service.impl;
3. import com.alatus.mall.order.entity.OrderEntity;
4. import com.alatus.mall.order.entity.OrderReturnReasonEntity;
5. import com.rabbitmq.client.Channel;
6. import org.springframework.amqp.core.Message;
7. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
9. import org.springframework.stereotype.Service;
10. import java.util.Map;
11. import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
12. import com.baomidou.mybatisplus.core.metadata.IPage;
13. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
14. import com.alatus.common.utils.PageUtils;
15. import com.alatus.common.utils.Query;
17. import com.alatus.mall.order.dao.OrderItemDao;
18. import com.alatus.mall.order.entity.OrderItemEntity;
19. import com.alatus.mall.order.service.OrderItemService;
22. @Service("orderItemService")
23. @RabbitListener(queues = "helloJava")
24. public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
25.     @Override
26.     public PageUtils queryPage(Map<String, Object> params) {
27.         IPage<OrderItemEntity> page = this.page(
28.                 new Query<OrderItemEntity>().getPage(params),
29.                 new QueryWrapper<OrderItemEntity>()
30.         );
32.         return new PageUtils(page);
33.     }
35.     @RabbitHandler
36.     public void getMessage(Message message, OrderReturnReasonEntity entity, Channel channel){
37.         System.out.println(entity);
38.     }
39.     @RabbitHandler
40.     public void getMessage1(OrderEntity entity){
41.         System.out.println(entity);
42.     }
43. }

复制代码

1. package com.alatus.mall.order.service.impl;
3. import com.alatus.mall.order.entity.OrderEntity;
4. import com.alatus.mall.order.entity.OrderReturnReasonEntity;
5. import com.rabbitmq.client.Channel;
6. import org.springframework.amqp.core.Message;
7. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
9. import org.springframework.stereotype.Service;
10. import java.util.Map;
11. import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
12. import com.baomidou.mybatisplus.core.metadata.IPage;
13. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
14. import com.alatus.common.utils.PageUtils;
15. import com.alatus.common.utils.Query;
17. import com.alatus.mall.order.dao.OrderItemDao;
18. import com.alatus.mall.order.entity.OrderItemEntity;
19. import com.alatus.mall.order.service.OrderItemService;
22. @Service("orderItemService")
23. @RabbitListener(queues = "helloJava")
24. public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
25.     @Override
26.     public PageUtils queryPage(Map<String, Object> params) {
27.         IPage<OrderItemEntity> page = this.page(
28.                 new Query<OrderItemEntity>().getPage(params),
29.                 new QueryWrapper<OrderItemEntity>()
30.         );
32.         return new PageUtils(page);
33.     }
35.     @RabbitHandler
36.     public void getMessage(Message message, OrderReturnReasonEntity entity, Channel channel){
37.         System.out.println(entity);
38.     }
39.     @RabbitHandler
40.     public void getMessage1(OrderEntity entity){
41.         System.out.println(entity);
42.     }
43. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。