当前位置：首页 > news >正文

上海网站建设与设计公司/带佣金的旅游推广平台有哪些

news 2025/7/5 17:06:25

上海网站建设与设计公司,带佣金的旅游推广平台有哪些,传奇网页游戏赤月,防止域名失效请牢记发布确认 1、发布确认原理生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产…

发布确认

1、发布确认原理

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了。如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm 模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。

2、单个发布确认

单个发布确认是一个同步的发布确认模式，生产者发布一条数据，等待broker的确认。确认完毕后，发布下一条。waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

优点：确保消息发送到broker之前，不会丢失，稳定性强

缺点：速度慢，同步发布确认，发一条，等待确认一条。

/*** 单个发布确认*/public static void publishComfirmIndividually() throws Exception {Channel channel = RabbitMQUtils.getChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);/*** 开启发布确认* 一旦模式开启，会为channel内的每一条消息，都配置上一个为一个的ID或者是标签名*/channel.confirmSelect();long start = System.currentTimeMillis();for (int i = 1; i <= MESSAGE_NUMBER; i++) {boolean flag = false;String message = "消息: " + i;flag = sendMessage(message, channel);while (!flag) {flag = sendMessage(message, channel);}}long end = System.currentTimeMillis();System.out.println("发布" + MESSAGE_NUMBER + "条消息，耗时：" + (end - start) + "ms");}public static boolean sendMessage(String message, Channel channel) throws Exception {channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));//等待确认return channel.waitForConfirms();}

3、批量发布确认

与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是**同步(只有确认成功才能继续发消息)**的，也一样阻塞消息的发布。

/*** 批量发布确认*/
public static void publishComfirmBatch() throws Exception {Channel channel = RabbitMQUtils.getChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);/*** 开启发布确认* 一旦模式开启，会为channel内的每一条消息，都配置上一个为一个的ID或者是标签名*/channel.confirmSelect();// 批量确认消息的 大小int batchSize = 100;// 为确认消息的个数int unconfirmedCount = 0;long start = System.currentTimeMillis();for (int i = 1; i <= MESSAGE_NUMBER; i++) {String message = "消息: " + i;channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));unconfirmedCount++;// 如果为确认的消息个数达到 指定的批量确认的条件,进行等待确认，要求broker给与确认if(unconfirmedCount == batchSize){boolean flag = channel.waitForConfirms();if(!flag){System.out.println("有确认失败的消息,但不知道是哪个");}unconfirmedCount = 0;}}//还有未被确认的，等待确认if(unconfirmedCount > 0){channel.waitForConfirms();}long end = System.currentTimeMillis();System.out.println("发布" + MESSAGE_NUMBER + "条消息，耗时：" + (end - start) + "ms");
}

4、异步发布确认

异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功，下面就让我们来详细讲解异步确认是怎么实现的。
在这里插入图片描述

/*** 异步发布确认*/
public static void publishComfirmAsync() throws Exception {Channel channel = RabbitMQUtils.getChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);/*** 开启发布确认* 一旦模式开启，会为channel内的每一条消息，都配置上一个为一个的ID或者是标签名或者是消息唯一的序列号*/channel.confirmSelect();/*** 什么是concurrentSkipListMap* 线程安全有序的一个哈希表，适用于高并发的情况* 1. 轻松的将序号和消息关联、* 2、轻松的批量删除条目，只需要给出序号* 3、支持并发访问*/ConcurrentSkipListMap<Long, String> unconfirmedMessageMap = new ConcurrentSkipListMap<>();/*** 确认收到一个消息的回调* 1. 发送的消息标志号，消息序列号* 2. true 可以确认小于等于当前序列号的消息（批量确认下）*    false 确认当前序列号的消息*/ConfirmCallback ackCallback = (deliverTag,multiple)->{if(multiple){//返回的是小于等于当前序列号的未确认消息 是一个 mapConcurrentNavigableMap<Long, String> confirmedMap = unconfirmedMessageMap.headMap(deliverTag,true);//清除该部分未确认消息confirmedMap.clear();}else{// 移除当前确认的 消息序号unconfirmedMessageMap.remove(deliverTag);}};// broker未收到消息的回调ConfirmCallback nackCallback = (deliverTag,multiple)->{// 通过 未被broker确认的消息序列号deliverTag，获取String message = unconfirmedMessageMap.get(deliverTag);System.out.println("未被broker确认的消息："+message);};// 为channel 增加发布确认监听器channel.addConfirmListener(ackCallback,nackCallback);long start = System.currentTimeMillis();for (int i = 1; i <= MESSAGE_NUMBER; i++) {String message = "消息: " + i;// getNextPublishSeqNo 返回下一个要被发布消息的序列号unconfirmedMessageMap.put(channel.getNextPublishSeqNo(),message);channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));}long end = System.currentTimeMillis();System.out.println("发布" + MESSAGE_NUMBER + "条消息，耗时：" + (end - start) + "ms");
}