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

织梦手机网站模板/模板建站常规流程

织梦手机网站模板,模板建站常规流程,有什么网站可以做批发,四川建设集团有限公司网站目录 消息应答 概念 自动应答 消息应答的方法 Multiple 的解释 手动应答实现 1.准备工具类 2.生产者 3.两个睡眠时间不同的消费者 4.效果展示 消息自动重新入队 效果演示: 消息应答 概念 消费者完成一个任务可能需要一段时间,如果其中一个消…

目录

消息应答

概念

自动应答

消息应答的方法

Multiple 的解释

手动应答实现

1.准备工具类

2.生产者

3.两个睡眠时间不同的消费者

4.效果展示

消息自动重新入队

 效果演示:


消息应答

概念

消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会导致消息丢失。RabbitMQ 一旦向消费者传递了一条消息,便立即将该消息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将丢失正在处理的消息。以及后续发送给该消费这的消息,因为它无法接收到。

为了保证消息在发送过程中不丢失,rabbitmq 引入消息应答机制,消息应答就是:消费者在接收到消息并且处理该消息之后,告诉 rabbitmq 它已经处理了,rabbitmq 可以把该消息删除了。

自动应答

消息发送后立即被认为已经传送成功

弊端:如果消息在接收到之前,消费者那边出现连接或者 信道 关闭,那么消息就丢 失了,当然另一方面这种模式消费者那边可以传递过载的消息,没有对传递的消息数量进行限制, 当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终 使得内存耗尽,最终这些消费者线程被操作系统杀死

所以在实际开发中我们应选择手动应答

消息应答的方法

Channel.basicAck():用于肯定确认

RabbitMQ 已知道该消息并且成功的处理消息,可以将其丢弃了

Channel.basicNack():用于否定确认

Channel.basicReject():用于否定确认 (推荐使用

Multiple 的解释

手动应答的好处是可以批量应答并且减少网络拥堵

 multiple 的 true 和 false 代表不同意思

        true 代表批量应答 channel 上未应答的消息

                比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8

                那么此时 5-8 的这些还未应答的消息都会被确认收到消息应答

       false 同上面相比

                只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答

手动应答实现

默认消息采用的是自动应答,所以我们要想实现消息消费过程中不丢失,需要把自动应答改 为手动应答

1.准备工具类

/*** 连接工厂创建信道的工具类*/
public class RabbitMQUtil {public static Channel getChannel() throws Exception {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();//设置工厂的IP,连接RabbitMQ的队列factory.setHost("192.168.31.65");//设置用户名factory.setUsername("admin");//密码factory.setPassword("123");//创建连接Connection connection = factory.newConnection();//获取信道return connection.createChannel();}
}
/*** 睡眠:模拟执行业务逻辑操作需要时间*/
public class SleepUtil {public static void sleep(int second) {try {Thread.sleep(1000 * second);} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}
}

2.生产者

/*** 消息在手动应答时不丢失,放回队列后重新消费*/
public class Producer {public static final String TASK_QUEUE_NAME = "ack_queue";public static void main(String[] args) throws Exception {Channel channel = RabbitMQUtil.getChannel();//开启发布确认channel.confirmSelect();//声明队列channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null);//从控制台中输入信息Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String message = scanner.next();channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("生产者发出消息" + message);}}
}

3.两个睡眠时间不同的消费者

/*** 消息在手动应答时不丢失,放回队列后重新消费*/
public class Worker01 {//队列的名称public static final String TASK_QUEUE_NAME = "ack_queue";//接受消息public static void main(String[] args) throws Exception {//获取信道Channel channel = RabbitMQUtil.getChannel();System.out.println("消费者一:等待接受___时间较短");DeliverCallback deliverCallback = (consumerTag, message) -> {//沉睡1秒,模拟执行代码需要一定的时间SleepUtil.sleep(1);System.out.println("接受到的消息" + new String(message.getBody()));//进行手动应答/** 1.消息的标记 tag,表明消息的唯一标识* 2.是否批量应答 应该不允许批量,防止消息的丢失* */channel.basicAck(message.getEnvelope().getDeliveryTag(), false);};//消息被取消时执行下面的内容CancelCallback cancelCallback = consumerTag -> {System.out.println(consumerTag + "消息消费被中断");};//消费者进行消息的消费channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, cancelCallback); //采用手动应答(false)}
}
/*** 消息在手动应答时不丢失,放回队列后重新消费*/
public class Worker02 {//队列的名称public static final String TASK_QUEUE_NAME = "ack_queue";//接受消息public static void main(String[] args) throws Exception {//获取信道Channel channel = RabbitMQUtil.getChannel();System.out.println("消费者二:等待接受___时间较长");DeliverCallback deliverCallback = (consumerTag, message) -> {//沉睡10秒,模拟执行代码需要一定的时间SleepUtil.sleep(10);System.out.println("接受到的消息" + new String(message.getBody()));//进行手动应答/** 1.消息的标记 tag,表明消息的唯一标识* 2.是否批量应答 应该不允许批量,防止消息的丢失* */channel.basicAck(message.getEnvelope().getDeliveryTag(), false);};//消息被取消时执行下面的内容CancelCallback cancelCallback = consumerTag -> {System.out.println(consumerTag + "消息消费被中断");};//消费者进行消息的消费channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, cancelCallback); //采用手动应答(false)}
}

4.效果展示

正常情况下消息发送方发送两个消息 消费者1 和 消费者2 分别接收到消息并进行处理

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或 TCP 连接丢失),导致消息 未发送 ACK 确认,RabbitMQ 将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者 可以处理,它将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确保不会丢失任何消息。 

 效果演示:

将上述代码中消费者2(睡眠时间更长)在消费消息的途中停止

发送者发送消息 ,发出消息之后的把 消费者2 停掉,按理说该 消费者2 来处理该消息,但是由于它处理时间较长,在还未处理完,也就是说 消费者2 还没有执行 ack 代码的时候,消费者2 被停掉了, 此时会看到消息被 消费者1 接收到了,说明消息 被重新入队,然后分配给能处理消息的 消费者1处理了

 

http://www.jmfq.cn/news/4806091.html

相关文章:

  • 呼和浩特市手机网站/推广普通话手抄报模板
  • 南山做网站的公司/网店无货源怎么做
  • 网站建设 利润/2024年新冠第三波症状分析
  • 企业网站建设资金预算表/郭生b如何优化网站
  • 沈阳手机网站制作/深圳seo外包公司
  • 下载百度导航app/佳木斯seo
  • 玉田建设局网站/网站推广网络营销
  • 网站接口怎么做/全球网站排行榜
  • 高端网站建设 骆诗/百度知道一下首页
  • 网站开发待遇怎么样/百度指数功能有哪些
  • 上海营销型企业网站/app网络推广公司
  • microsoft做网站/网络营销整合推广
  • 网站关键词优化排名技巧/推广平台网站有哪些
  • 辽宁建设工程信息网查询系统/短视频seo公司
  • 网站开发ppt转h5/南京搜索引擎推广优化
  • 专门做定制化的网站/网页制作代码
  • 做网站服务器软件/百度搜索关键词排名优化技术
  • 做网站满屏的照片尺寸是多少/交换神器
  • 做网站好的网络公司/汕头seo网站建设
  • 营销型网站建设需要多少钱/湖南疫情最新消息
  • 济宁网站建设公司公司/最新nba排名
  • 网站建设与管理用什么软件有哪些/为什么打开网址都是站长工具
  • 营销型网站有哪些出名的/今日最新新闻
  • 小米的网站建设的要点/app开发定制
  • 南阳建设网站制作/aso优化软件
  • 网络舆情监测处置制度/惠州seo网站推广
  • 宝鸡网站建设公司/深圳网络营销怎么推广
  • jsp环保主题网站代做/app推广拉新平台
  • 做网站说什么5.0啥意思/网站优化排名提升
  • 中企动力网站模板/商城网站建设