转：/u014373554/article/details/92686063
项目是使用springboot项目开发的，前是代码实现，后面有分析发送消息失败、消息持久化、消费者失败处理方法和发送消息解决方法及手动确认的模式
先引入pom.xml
<!--rabbitmq--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
application 配置文件
spring:rabbitmq:host: IP地址port: 5672username: 用户名password: 密码RabbitConfig配置文件import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.DirectExchange;import org.springframework.amqp.core.Queue;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.support.converter.Jackson2JsonMessageConverter;import org.springframework.beans.factory.annotation.Value;import org.springframework.beans.factory.config.ConfigurableBeanFactory;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.context.annotation.Scope;/**Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。Queue:消息的载体,每个消息都会被投到一个或多个队列。Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.Routing Key:路由关键字,exchange根据这个关键字进行消息投递。vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。Producer:消息生产者,就是投递消息的程序.Consumer:消息消费者,就是接受消息的程序.Channel:消息通道,在客户端的每个连接里,可建立多个channel.*/@Configuration@Slf4jpublic class RabbitConfig {@Value("${spring.rabbitmq.host}")private String host;@Value("${spring.rabbitmq.port}")private int port;@Value("${spring.rabbitmq.username}")private String username;@Value("${spring.rabbitmq.password}")private String password;public static final String EXCHANGE_A = "my_mq_exchange_A";public static final String EXCHANGE_B = "my_mq_exchange_B";public static final String EXCHANGE_C = "my_mq_exchange_C";public static final String QUEUE_A="QUEUE_A";public static final String QUEUE_B="QUEUE_B";public static final String QUEUE_C="QUEUE_C";public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";public static final String ROUTINGKEY_B = "spring-boot-routingKey_B";public static final String ROUTINGKEY_C = "spring-boot-routingKey_C";@Beanpublic ConnectionFactory connectionFactory(){CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);connectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setVirtualHost("/");connectionFactory.setPublisherConfirms(true); //设置发送消息失败重试connectionFactory.setChannelCacheSize(100);//解决多线程发送消息return connectionFactory;}@Bean@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)public RabbitTemplate rabbitTemplate(){RabbitTemplate template = new RabbitTemplate(connectionFactory());template.setMandatory(true); //设置发送消息失败重试return template;}//配置使用json转递数据@Beanpublic Jackson2JsonMessageConverter producerJackson2MessageConverter() {return new Jackson2JsonMessageConverter();}/*public SimpleMessageListenerContainer messageListenerContainer(){SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageHandler());adapter.setDefaultListenerMethod(new Jackson2JsonMessageConverter());return container;}*//*** 针对消费者配置* 1. 设置交换机类型* 2. 将队列绑定到交换机* FanoutExchange: 将消息分发到所有的绑定队列，无 routingkey的概念* HeadersExchange: 通过添加属性key - value匹配* DirectExchange: 按照routingkey分发到指定队列* TopicExchange : 多关键字匹配* @return*/@Beanpublic DirectExchange defaultExchange(){return new DirectExchange(EXCHANGE_A,true,false);}@Beanpublic Queue queueA(){return new Queue(QUEUE_A,true);// 队列持久化}@Beanpublic Queue queueB(){return new Queue(QUEUE_B,true);// 队列持久化}/*** 一个交换机可以绑定多个消息队列，也就是消息通过一个交换机,可以分发到不同的队列当中去。* @return*/@Beanpublic Binding binding(){return BindingBuilder.bind( queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);}@Beanpublic Binding bindingB(){return BindingBuilder.bind( queueB()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);}}
生成者
import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.amqp.rabbit.support.CorrelationData;import org.springframework.beans.factory.annotation.Autowired;import org.ponent;import java.util.UUID;/*** 生产者*/@Component@Slf4jpublic class ProducerMessage implements RabbitTemplate.ConfirmCallback , RabbitTemplate.ReturnCallback{private RabbitTemplate rabbitTemplate;@Autowiredpublic ProducerMessage(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;rabbitTemplate.setConfirmCallback(this::confirm); //rabbitTemplate如果为单例的话，那回调就是最后设置的内容rabbitTemplate.setReturnCallback(this::returnedMessage);rabbitTemplate.setMandatory(true);}public void sendMsg (Object content){CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A,RabbitConfig.ROUTINGKEY_A,content,correlationId);}/*** 消息发送到队列中，进行消息确认* @param correlationData* @param ack* @param cause*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {log.info(" 消息确认的id： " + correlationData);if(ack){log.info("消息发送成功");//发送成功 删除本地数据库存的消息}else{log.info("消息发送失败：id "+ correlationData +"消息发送失败的原因"+ cause);// 根据本地消息的状态为失败，可以用定时任务去处理数据}}/*** 消息发送失败返回监控* @param message* @param i* @param s* @param s1* @param s2*/@Overridepublic void returnedMessage(Message message, int i, String s, String s1, String s2) {log.info("returnedMessage [消息从交换机到队列失败] message："+message);}}
消费者
import com.rabbitmq.client.Channel;import lombok.extern.slf4j.Slf4j;import net.sf.json.JSONObject;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.ponent;import java.io.IOException;/*** 消费者*/@Slf4j@Componentpublic class ComsumerMessage {@RabbitListener(queues = RabbitConfig.QUEUE_A)public void handleMessage(Message message,Channel channel) throws IOException{try {String json = new String(message.getBody());JSONObject jsonObject = JSONObject.fromObject(json);log.info("消息了【】handleMessage" + json);int i = 1/0;//业务处理。/*** 防止重复消费，可以根据传过来的唯一ID先判断缓存数据中是否有数据* 1、有数据则不消费，直接应答处理* 2、缓存没有数据，则进行消费处理数据，处理完后手动应答* 3、如果消息 处理异常则，可以存入数据库中，手动处理（可以增加短信和邮件提醒功能）*///手动应答channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}catch (Exception e){log.error("消费消息失败了【】error："+ message.getBody());log.error("OrderConsumer handleMessage {} , error:",message,e);// 处理消息失败，将消息重新放回队列channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);}}}
发送消息：调用生成的方法
import com.zz.blog.BlogApplicationTests;import com.zz.blog.mq.ProducerMessage;import net.sf.json.JSONObject;import org.junit.Test;import org.springframework.beans.factory.annotation.Autowired;import java.util.UUID;public class Message extends BlogApplicationTests {@Autowiredprivate ProducerMessage producerMessage;@Testpublic void sendMessage(){JSONObject jsonObject = new JSONObject();jsonObject.put("id", UUID.randomUUID().toString());jsonObject.put("name","TEST");jsonObject.put("desc","订单已生成");//防止发送消息失败，将发送消息存入本地。producerMessage.sendMsg(jsonObject.toString());}}
rabbitTemplate的发送消息流程是这样的：
1 发送数据并返回(不确认rabbitmq服务器已成功接收)
2 异步的接收从rabbitmq返回的ack确认信息
3 收到ack后调用confirmCallback函数
注意：在confirmCallback中是没有原message的，所以无法在这个函数中调用重发，confirmCallback只有一个通知的作用
在这种情况下，如果在2，3步中任何时候切断连接，我们都无法确认数据是否真的已经成功发送出去，从而造成数据丢失的问题。
最完美的解决方案只有1种：
使用rabbitmq的事务机制。
但是在这种情况下，rabbitmq的效率极低，每秒钟处理的message在几百条左右。实在不可取。
基于上面的分析，我们使用一种新的方式来做到数据的不丢失。
在rabbitTemplate异步确认的基础上
1 在本地缓存已发送的message
2 通过confirmCallback或者被确认的ack，将被确认的message从本地删除
3 定时扫描本地的message，如果大于一定时间未被确认，则重发
当然了，这种解决方式也有一定的问题：
想象这种场景，rabbitmq接收到了消息，在发送ack确认时，网络断了，造成客户端没有收到ack，重发消息。（相比于丢失消息，重发消息要好解决的多，我们可以在consumer端做到幂等）。
消息存入本地：在message 发消息的写数据库中。
消息应答成功，则删除本地消息，失败更改消息状态，可以使用定时任务去处理。
消息持久化：
消费者：
/*** 防止重复消费，可以根据传过来的唯一ID先判断缓存数据库中是否有数据* 1、有数据则不消费，直接应答处理* 2、缓存没有数据，则进行消费处理数据，处理完后手动应答* 3、如果消息 处理异常则，可以存入数据库中，手动处理（可以增加短信和邮件提醒功能）*/
