项目是使用springboot项目开发的，前是代码实现，后面有分析发送消息失败、消息持久化、消费者失败处理方法和发送消息解决方法及手动确认的模式

先引入pom.xml

<!--rabbitmq-->

<dependency>

   <groupId>org.springframework.boot</groupId>

   <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

application 配置文件

spring:

rabbitmq:

  host: IP地址

  port: 5672

  username: 用户名

  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

@Slf4j

public 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";

    @Bean

    public 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转递数据

    @Bean

    public 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

     */

    @Bean

    public DirectExchange defaultExchange(){

        return new DirectExchange(EXCHANGE_A,true,false);

    }

    @Bean

    public Queue queueA(){

        return  new Queue(QUEUE_A,true);// 队列持久化

    }

    @Bean

    public Queue queueB(){

        return  new Queue(QUEUE_B,true);// 队列持久化

    }

    /**

     * 一个交换机可以绑定多个消息队列，也就是消息通过一个交换机,可以分发到不同的队列当中去。

     * @return

     */

    @Bean

    public Binding binding(){

        return BindingBuilder.bind( queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);

    }

    @Bean

    public 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.springframework.stereotype.Component;

import java.util.UUID;

/**

 * 生产者

 */

@Component

@Slf4j

public class ProducerMessage implements  RabbitTemplate.ConfirmCallback , RabbitTemplate.ReturnCallback{

    private RabbitTemplate rabbitTemplate;

    @Autowired

    public 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

     */

    @Override

    public 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

     */

    @Override

    public 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.springframework.stereotype.Component;

import java.io.IOException;

/**

 * 消费者

 */

@Slf4j

@Component

public 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 {

    @Autowired

    private ProducerMessage producerMessage;

    @Test

    public 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、如果消息 处理异常则，可以存入数据库中，手动处理（可以增加短信和邮件提醒功能）

 */