一、数据丢失
第一种：生产者弄丢了数据。生产者将数据发送到 RabbitMQ 的时候，可能数据就在半路给搞丢了，因为网络问题，都有可能。
第二种：RabbitMQ 弄丢了数据。MQ还没有持久化自己挂了。
第三种：消费端弄丢了数据。刚消费到，还没处理，结果进程挂了，比如重启了。
解决方案
消费者从队列中获取到数据，应答成功之后，队列认为消费者对消息处理完成，从队列中删除消息。
生产者
mq告知生产者，消息接收成功，没有成功，生产者可以继续重新发送消息
RabbitMQ有两种方式来解决这个问题：
-
通过AMQP提供的事务机制实现（同步，不推荐）；
-
使用发送者确认模式实现（confirm，异步）；
事务使用
事务的实现主要是对信道（Channel）的设置，主要的方法有三个：
-
channel.txSelect()声明启动事务模式；
-
channel.txCommit()提交事务；
-
channel.txRollback()回滚事务；
public class Product {
public static void main(String[] args) throws IOException, TimeoutException {Connection conn = ConnUtils.getConn();Channel channel = conn.createChannel();
String queueName = "my-que";// queueDeclare(String queue队列名, boolean durable持久化, boolean exclusive是否排外, boolean autoDelete自动删除,Map<String, Object> arguments队列其他属性信息)channel.queueDeclare(queueName,false,false,false,null);
String message = "hello rabbitmq";
try{// 开启事务channel.txSelect();// 发布消息channel.basicPublish("",queueName,null,message.getBytes());// 提交事务channel.txCommit();}catch (Exception e){e.printStackTrace();// 回滚事务channel.txRollback();}channel.close();conn.close();}
}
Confirm发送方确认模式
Confirm发送方确认模式使用和事务类似，也是通过设置Channel进行发送方确认的。
Confirm的三种实现方式：
-
channel.waitForConfirms()普通发送方确认模式，单条数据；
-
channel.waitForConfirmsOrDie()批量确认模式，多条数据；
-
channel.addConfirmListener()异步监听发送方确认模式；
public class Product {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection conn = ConnUtils.getConn();// 创建信道：处理消息Channel channel = conn.createChannel();
// 创建队列：存储消息String queueName = "my-que";// queueDeclare(String queue队列名, boolean durable持久化, boolean exclusive是否排外, boolean autoDelete自动删除,Map<String, Object> arguments队列其他属性信息)channel.queueDeclare(queueName, false, false, false, null);
// 定义消息String message = "hello rabbitmq";
// 开启发送方确认模式channel.confirmSelect();// basicPublish(交换器，序列名，参数信息，消息)channel.basicPublish("", queueName, null, message.getBytes());
if(channel.waitForConfirms()){System.out.println("发送成功");}
channel.close();conn.close();}
}
// 开启发送方确认模式
channel.confirmSelect();
for (int i = 0; i < 10; i++) {channel.basicPublish("", queueName, null, message.getBytes());
}
// 直到所有信息都发布，只要有一个未确认就会IOException
channel.waitForConfirmsOrDie();
System.out.println("全部执行完成");
// 开启发送方确认模式
channel.confirmSelect();
for (int i = 0; i < 10; i++) {channel.basicPublish("", queueName, null, message.getBytes());
}
//异步监听确认和未确认的消息
channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {System.out.println("未确认消息，标识：" + deliveryTag);}@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {System.out.println(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));}
});
System.out.println("执行操作...");
MQ
rabbitmq做持久化，持久化主要分为：
-
交换器的持久化
// 参数1 exchange ：交换器名 // 参数2 type ：交换器类型 // 参数3 durable ：是否持久化 channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
-
队列持久化
// 参数1 queue ：队列名 // 参数2 durable ：是否持久化 // 参数3 exclusive ：仅创建者可以使用的私有队列，断开后自动删除 // 参数4 autoDelete : 当所有消费客户端连接断开后，是否自动删除队列 // 参数5 arguments channel.queueDeclare(QUEUE_NAME, true, false, false, null);
-
消息持久化
// 参数1 exchange ：交换器 // 参数2 routingKey ： 路由键 // 参数3 props ： 消息的其他参数,其中 MessageProperties.PERSISTENT_TEXT_PLAIN 表示持久化 // 参数4 body ： 消息体 channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
消费者
关闭自动应答，改为手动应答，处理完成，MQ中再做删除
// 消费者DefaultConsumer consumer = new DefaultConsumer(channel) {@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("deliveryTag:" + envelope.getDeliveryTag());Thread.sleep(1000);System.out.println("获取到的消息内容:" + new String(body));// 手动应答channel.basicAck(envelope.getDeliveryTag(),false);}};
// false:关闭自动应答channel.basicConsume(queueName, false, consumer);
二、死信队列
-
消息过期会自动删除
-
消费者立刻接收消息进行处理
消费者立刻接收到数据处理，需求队列得数据不立刻被消费者处理，而是希望等待一段时间才被处理
死信队列：是延迟队列得一种实现方式，不是一种特殊得队列，死信交换机+消息有效期
-
交换机的类型：fanout，direct，topic，header
-
队列：不区分类型，也是一个普通得队列
-
死信交换机：将一个交换机设置为某一个队列的死信交换机，当前队列过期的消息会发送给交换机
-
消息：有效期
消息传递顺序
生产者发布消息 --> 交换机(转发消息) --> 队列(接收存储消息) --> 消费者(接收处理消息)
队列转发消息：
-
正常消息：转给消费者处理；
-
过期消息：默认删除，也可以通过配置（死信交换机）转发给交换机；
交换机的消息来源：
-
生产者发布的（普通得生产者）
-
其他队列到期删除的消息自动进入队列（队列需要绑定交换机，队列发布）
队列指的内容：
-
消息来源都是从交换机发送得
死信队列：交换机B作为队列A的死信交换机，接收队列A中被删除的消息，实现队列A中的消息延迟处理的效果
死信队列DLX（dead-letter-exchange），利用DLX，当消息在一个队列中变成死信 (dead message) 之后，它能被重新投递到另一个exchange上，这个exchange就是DLX，似乎叫死信交换机更加贴切，当死信投递到这个exchange后，我们也可以用一个queue来绑定该exchange，该exchange就可以根据路由规则把这个死信路由到这个queue了，同样可以创建消费端去进行消费，以便对死信进行相应的处理。
延时队列：实际是不存在直接可用的延时队列，可通过死信消息和死信队列来实现延时队列的功能。RabbitMQ中的所有队列，消息存进来，立刻分发给对应的消费者进行处理。
死信交换机： DLX 全称（Dead-Letter-Exchange）。其实它是个普通的交换机，但它是设置在队列上某个参数的值对应的交换机。
死信队列：如果某个队列上存在参数：x-dead-letter-exchange， 当这个队列里的消息变成死信消息（dead message）后会被重新Pushlish到 x-dead-letter-exchange 所对应参数值的交换机上，跟这个交换机所绑定的队列就是死信队列。
为什么使用延迟队列
订单五分钟内支付：未支付订单超时
以前：1.用户做支付时，判断是否超时，修改订单状态；（不调用）
2.定时任务，每一秒刷新一次；（频繁刷新）
延时队列的应用场景很多，在我的项目开发中也涉及到很多，例如：订单五分钟未支付自动取消、订单准备超时30分钟推送提醒给门店、订单完成后两小时推送评价邀请给用户等等，这些间隔指定时间后的操作都可以使用延时队列。
消息变成死信的情况
死信消息：
-
消息被拒绝(basic.reject / basic.nack)，并且requeue = false（不重回队列）。
-
消息TTL过期。
-
队列达到最大长度。
过期消息：RabbitMq 有两种设置消息过期的方式：
-
创建队列时通过 x-message-ttl 参数指定该队列消息的过期时间，这种队列里的消息过期时间全部相同。
-
生产者Pushlish消息时，通过设置消息的 expiration 参数指定过期时间，每个消息的过期时间都不一样。
如果两者同时使用，过期时间按照小的一方为准，两种方式设置的时间都是 毫秒。
Queue参数信息
-
name: 队列的名称；
-
actualName: 队列的真实名称，默认用name参数，如果name为空，则根据规则生成一个；
-
durable: 是否持久化；
-
exclusive: 是否独享、排外的；
-
autoDelete: 是否自动删除；
-
arguments：队列的其他属性参数，有如下可选项，可参看图2的arguments：
-
x-message-ttl：消息的过期时间，单位：毫秒；
-
x-expires：队列过期时间，队列在多长时间未被访问将被删除，单位：毫秒；
-
x-max-length：队列最大长度，超过该最大值，则将从队列头部开始删除消息；
-
x-max-length-bytes：队列消息内容占用最大空间，受限于内存大小，超过该阈值则从队列头部开始删除消息；
-
x-overflow：设置队列溢出行为。这决定了当达到队列的最大长度时消息会发生什么。有效值是drop-head、reject-publish或reject-publish-dlx。仲裁队列类型仅支持drop-head；
-
x-dead-letter-exchange：死信交换器名称，过期或被删除（因队列长度超长或因空间超出阈值）的消息可指定发送到该交换器中；
-
x-dead-letter-routing-key：死信消息路由键，在消息发送到死信交换器时会使用该路由键，如果不设置，则使用消息的原来的路由键值
-
x-single-active-consumer：表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，其他被忽略，false时消息循环分发给所有消费者(默认false)
-
x-max-priority：队列要支持的最大优先级数;如果未设置，队列将不支持消息优先级；
-
x-queue-mode（Lazy mode）：将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用;如果未设置，队列将保留内存缓存以尽可能快地传递消息；
-
x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。
-
交换机参数信息
类型
-
FanoutExchange ：扇形 交换机
-
DirectExchange ：直连交换机 routing模式用的交换机
-
TopicExchange ：主题模式交换机
参数信息
-
name：交换机名称
-
durable：是否把交换机持久化到磁盘上
-
autoDelete ：是否自动删除交换机 用法有点类似上面的队列中的autoDelete 只有所有队列都和交换机接触订阅，说白了就是所有绑定到交换机上的队列不再需要改交换机，他就该死了,有点残酷
-
artuments ：额外参数
实现
生产者
发布消息，设置消息过期时间
public class Producer {
// default exchangeprivate static String exchange = "test_dlx_exchange";// default exchange 的路由规则： routingKey（test） 将匹配同名的 queue(test)private static String routingKey = "dlx.abc";
public static void main(String[] args) throws IOException, TimeoutException {
// 1 创建ConnectionConnection connection = ConnUtils.getConn();
// 2 创建ChannelChannel channel = connection.createChannel();
// 3 发送消息for (int i = 0; i < 5; i++) {AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()// deliveryMode:1（nopersistent）非持久化，2（persistent）持久化.deliveryMode(2).contentEncoding("UTF-8")// expiration:消息的失效时间.expiration("50000").build();String msg = "RabbitMQ: dlx message" + i;channel.basicPublish(exchange , routingKey , properties , msg.getBytes());}
// 5 关闭连接channel.close();connection.close();}
}
正常消费者
绑定死信队列，当消息变为死信消息时，会自动存入私信队列
public class Consumer {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnUtils.getConn();
Channel channel = connection.createChannel();
// 定义死信队的ExchangeString dlxExchange = "dlx.exchange";channel.exchangeDeclare(dlxExchange, "topic");
// 死信队列名String dlxQueue = "dlx.queue";channel.queueDeclare(dlxQueue, true, false, false, null);
// # 表示所有的key都可以路由到s死信队列String dlxRoutingKey = "#";// 绑定死信队列和exchangechannel.queueBind(dlxQueue, dlxExchange, dlxRoutingKey, null);
// 定义正常的消费者j监听队列// 申明exchangeString exchangeName = "test_dlx_exchange";channel.exchangeDeclare(exchangeName, "topic");
// 申明队列Map<String, Object> arguments = new HashMap<>();// 设置死信队列,arguments要设置到申明的队列上arguments.put("x-dead-letter-exchange", dlxExchange);String queueName = "test_dlx_queue";channel.queueDeclare(queueName, true, false, false, arguments);
// 队列绑定到exchangeString routingKey = "dlx.#";channel.queueBind(queueName, exchangeName, routingKey);
channel.basicQos(1);
DeliverCallback deliverCallback = new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("---消费者-- " + new String(message.getBody(), "UTF-8"));
}};
CancelCallback cancelCallback = new CancelCallback() {@Overridepublic void handle(String consumerTag) throws IOException {System.out.println("---消费者--：cancelCallback ");}};
// callback: 消费者会调函数，处理发送过来的消息。// cancelCallback: 消费者取消订阅时的回调方法。// 消费消息,autoAck一定要设为false，手工ackchannel.basicConsume(queueName, false, deliverCallback, cancelCallback);}
}
死信消费者
处理死信队列中得信息
// DLXConusmer,监听消费死信队列中的消息
public class DLXConusmer {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnUtils.getConn();
Channel channel = connection.createChannel();
String queueName = "dlx.queue";String exchangeName = "dlx.exchange";String routingKey = "#";
// 申明exchangechannel.exchangeDeclare(exchangeName, "topic");// 申明队列channel.queueDeclare(queueName, true, false, false, null);// 队列绑定到exchangechannel.queueBind(queueName, exchangeName, routingKey, null);
DeliverCallback deliverCallback = new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {try {System.out.println("---死信队列消费者---");System.out.println(new String(message.getBody(), "UTF-8"));} finally {// consumer手动 ack 给brokerchannel.basicAck(message.getEnvelope().getDeliveryTag(), false);}}};CancelCallback cancelCallback = new CancelCallback() {@Overridepublic void handle(String consumerTag) throws IOException {System.out.println("---消费者：cancelCallback");}};
// 消费消息,autoAck一定要设置为falsechannel.basicConsume(queueName, false, deliverCallback, cancelCallback);}
}
