一、确保消费者customer处理消息成功
默认情况下消费者C1接收到消息1无论是否正常接受和处理都会立即应答rabbit服务器，然后消息1就会从队列中被删除，假如C1突然出现异常状况导致消息1没有被处理完毕，那么消息1就处理失败了，也不会有其他消费者去处理消息1。事实上我们希望的是消息1如果没有被C1正确处理完毕，那么就发送给其他消费者处理，为了达到这个目的，只需要做两件事情，第一关闭rabbitMq的自动应答机制，第二消费者正确处理完消息后手动应答。
RabbitMQ应答机制：
- 自动确认，默认是自动确认，即获取消息后，直接确认。
- 手动确认，给当前消息设置状态，当手动ack后服务端才会删除该消息，如果返回nack，重新入队。
customer在监听队列接收消息的时候，申明取消自动应答，手动返回完成。
channel.basicConsume(QUEUE_NAME, false, consumer);
在完成消费操作时，返回确认状态。
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
- delivery.getEnvelope().getDeliveryTag(): 消息id
- true: 这里true或者false都代表已经应答
二、存在两个大问题
问题： 1）重复消费 2） 消息堆积
1. 消息重发导致消息重复消费的问题（消息中间件幂等性）
如果在消费方customer在完成消费了之后，由于网络问题没有及时应答，就会存在大量消息堆积在MQ服务器。
由于RabbitMQ有消息重新发送的机制，如果没有及时回应那么就会继续重发，重发就会导致消息重复消费。（1）在生产者producer产生消息的时候可以给消息一个唯一的id。
（2）在执行完毕之后，利用redis缓存消息id，判断时候消费过。单个消费者：
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String value = redis.get("key_"+envelope.getDeliveryTag()); if (value != null){ //之前已经执行过了，所以直接应答 channel.basicAck(deliveryTag, true); return ; } String exchange = envelope.getExchange();//交换 long deliveryTag = envelope.getDeliveryTag();//消息id String routingKey = envelope.getRoutingKey();//路由key String message = new String(body, "utf-8"); System.out.println(message); //先在redis中放入消息id，记得加上过期时间 // 返回确认状态} 多个消费者： 分布式锁解决
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try{ //在赋值锁的时候可以加上过期时间 boolean flag = redisTemplate.opsForValue().setIfAbsent("key_"+envelope.getDeliveryTag(),envelope.getDeliveryTag()); //如果没有被赋值则返回true if(flag){ String exchange = envelope.getExchange();//交换 long deliveryTag = envelope.getDeliveryTag();//消息id String routingKey = envelope.getRoutingKey();//路由key String message = new String(body, "utf-8"); System.out.println(message); // 返回确认状态 } }catch(Exception e){ //打印日志 //删除redis中的记录 //直接return;不应答，等待再次重新发送。 }} 2. 消息堆积解决
1）加大rabbitMQ的内存空间
2）重启服务