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

常用ppt模板网站/能打开的a站

news 2025/8/8 21:03:00
常用ppt模板网站,能打开的a站,设计制作费税率,网站开发需要哪些文档RabbitMQ1、消息100%的投递2、幂等性概念3、Confirm确认消息4、Return返回消息5、自定义消费者1、消息100%的投递 消息如何保障100%的投递成功？ 什么是生产端的可靠性投递？ 保障消息的成功发出保障MQ节点的成功接收发送端收到MQ节点（Broker&…

RabbitMQ

  • 1、消息100%的投递
  • 2、幂等性概念
  • 3、Confirm确认消息
  • 4、Return返回消息
  • 5、自定义消费者

1、消息100%的投递

在这里插入图片描述
消息如何保障100%的投递成功？
什么是生产端的可靠性投递？

  • 保障消息的成功发出
  • 保障MQ节点的成功接收
  • 发送端收到MQ节点（Broker）确认应答
  • 完善的消息进行补偿机制
  • 微信支付一定会成功，发出消息
  • MQ服务器 能够收到支付微服务服务器发来的消息
  • 支付微服务 能够收到MQ服务器发来的确认应答的消息
  • 前面三步可能因为各种原因失败，就需要亡羊补牢，这也就是消息中间件所讲的 补偿机制

BAT/TMD互联网大厂的解决方案：
方法一：消息落库，对消息状态进行打标
方法二： 消息的延迟投递，做二次确认，回调检查
在这里插入图片描述
方法一
流程步骤：
流程如下：
第1步：将订单入库，创建一条MSG(状态为0) 入MSG DB库
第2步：将消息发出去
第3步：监听消息应答(来自Broker)
第4步：修改消息的状态为1(成功)
第5步：分布式定时任务抓取状态为0的消息
第6步：将状态为0的消息重发
第7步：如果尝试了3次(可按实际情况修改)以上则将状态置为2(消息投递失败状态)

5，6，7步为补偿机制
继续补偿：对于状态为2的消息，手动将对应的业务归档
在这里插入图片描述

方法二
第1步：首先业务数据落库，成功后才第一次消息发送
第2步：紧接着发送第2条消息（可以用于寻找第1条消息），用于延迟(可能2,3分钟后才发送)消息投递检查
第3步：Broker端收到消息后，消费端进行消息处理
第4步：处理成功后，发送confirm消息
第5步：收到confirm消息后，将消息进行持久化存储
第6步：收到了delay消息，检查DB数据库，若对应的第1条消息已处理完成，则不做任何事情；若收到了delay消息，检查DB数据库，发现对应的第1条消息处理失败(或无记录)，则发送重传命令到上游服务，循环第1步

第6步为补偿机制

2、幂等性概念

幂等性是什么？

  • 我们可以借鉴数据库的乐观锁机制
  • 比如我们执行一条更新库存的SQL语句
  • Update t_repository set count = count -1,version = version + 1 where version = 1
  • Elasticsearch也是严格遵循幂等性概念，每次数据更新，version+1

消费端-幂等性保障
在海量订单产生的业务高峰期，如何避免消息的重复消费问题？

消费实现幂等性，就意味着，我们的消息永远不会消费多次，即使我们收到了多条一样的消息

业界主流的幂等性操作
唯一ID+指纹码机制，利用数据库主键去重
利用Redis的原子性去实现

唯一ID+指纹码 机制
唯一ID+指纹码机制，利用数据库主键去重
Select count(1) from T_order where ID=唯一ID+指纹码
好处：实现简单
坏处：高并发下有数据库写入的性能瓶颈
解决方案：根据ID进行分库分表进行算法路由

利用Redis的原子性去实现
使用Redis进行幂等，需要考虑的问题
第一：我们是否要进行数据落库，如果落库的话，关键解决的问题是数据库和缓存如何做到原子性？
第二：如果不进行落库，那么都存储到缓存中，如何设置定时同步策略？

3、Confirm确认消息

理解Confirm消息确认机制
消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者 一个应答。
生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，这种方式也是消息的可靠性投递的核心保障
在这里插入图片描述
如何实现Confirm确认消息？
第一步：在Channel上开启确认模式：channel.confirmSelect()
第二步：在channel上添加监听：addConfirmListener，监听成功和失败的返回结果，根据具体的结果对消息进行重新发送、或记录日志等后续处理！

消费端

package com.myy.rabbitmqapi.confirm;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 13:54** 短信微服务*/
public class Consumer {public static void main(String[] args)  throws Exception{
//        1.创建ConnectFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.59.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//        2.获取ConnectionConnection connection = connectionFactory.newConnection();//        3.通过Connection创建一个新的ChannelChannel channel = connection.createChannel();String exchangeName = "test_confirm_exchange";String routingKey = "confirm.#";String queueName = "test_confirm_queue";//        4.声明交换机和队列 然后进行绑定设置, 最后制定路由Keychannel.exchangeDeclare(exchangeName, "topic", true);channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, exchangeName, routingKey);//        5.创建消费者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);channel.basicConsume(queueName, true, queueingConsumer);while(true){QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消费端: " + msg);}}
}

服务提供方

package com.myy.rabbitmqapi.confirm;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 13:58** 支付微服务*/
public class Producer {public static void main(String[] args) throws Exception {
//        1.创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.59.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//        2.获取ConnectionConnection connection = connectionFactory.newConnection();//        3.通过Connection创建一个新的ChannelChannel channel = connection.createChannel();//        4.指定我们的消息投递模式: 消息的确认模式channel.confirmSelect();String exchangeName = "test_confirm_exchange";String routingKey = "confirm.save";//        5.发送一条消息String msg = "Hello RabbitMQ Send confirm message!";channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());//        6.添加一个确认监听channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {System.err.println("-------no ack!-----------");}@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {System.err.println("----支付微服务---ack!-----------");}});}
}

运行消费端创建了队列
在这里插入图片描述

在这里插入图片描述
运行服务提供方接收
在这里插入图片描述
在这里插入图片描述

运行RabbitMQ时如果出现WARNING: IPv4 forwarding is disabled. Networking will not work可看这条博客解决问题

4、Return返回消息

Return Listener用于处理一些不可路由的消息！

  • 正常情况：我们的消息生产者，通过指定一个Exchange和RoutingKey，把消息送达到某一个队列中去，然后我们的消费者监听队列，进行消费处理操作！
  • 异常情况：在某些情况下，如果我们在发送消息的时候，当前的Exchange不存在或者指定的路由key路由不到，这个时候如果我们需要监听这种不可达的消息，就需要使用Return Listener！

在基础API中有一个关键的配置项
Mandatory：如果为true，则监听器会接收到路由不可达的消息，然后进行后续处理，如果为false，那么Broker端自动删除该消息！
在这里插入图片描述
消费端

package com.myy.rabbitmqapi.returnlistener;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 15:33*/
public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.59.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();String exchangeName = "test_return_exchange";String routingKey = "return.#";String queueName = "test_return_queue";channel.exchangeDeclare(exchangeName, "topic", true, false, null);channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, exchangeName, routingKey);QueueingConsumer queueingConsumer = new QueueingConsumer(channel);channel.basicConsume(queueName, true, queueingConsumer);while(true){QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消费者: " + msg);}}
}

生产端

package com.myy.rabbitmqapi.returnlistener;import com.rabbitmq.client.*;import java.io.IOException;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 15:32*/
public class Producer {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.59.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();String exchange = "test_return_exchange";String routingKey = "return.save";String routingKeyError = "abc.save";String msg = "Hello RabbitMQ Return Message";channel.addReturnListener(new ReturnListener() {@Overridepublic void handleReturn(int replyCode, String replyText, String exchange,String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {System.err.println("---------handle  return----------");System.err.println("replyCode: " + replyCode);System.err.println("replyText: " + replyText);System.err.println("exchange: " + exchange);System.err.println("routingKey: " + routingKey);System.err.println("properties: " + properties);System.err.println("body: " + new String(body));}});//消息投递成功，会被消费者所消费
//        channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());//消息不可达，将触发ReturnListenerchannel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());}
}

运行消费端
在这里插入图片描述
运行生产端
采用正确的routingKey，消息发送正确
在这里插入图片描述
在这里插入图片描述
再次运行消费端
采用错误的routingKey，此时没有指定的队列，需要在微信微服务中添加监听Return Listenter
在这里插入图片描述
在这里插入图片描述

回复的id,原因没有交换机，携带的详细内容都有显示

5、自定义消费者

我们一般就是在代码中编写while循环，进行consumer.nextDelivery方法进行获取下一条消息，然后进行消费处理！
但是我们使用自定义的Consumer更加的方便，解耦性更加的强，也是实际工作中最常用的使用方式！
之前：线程一直在等待生产端在发送消息，是不好的
自定义消费端

package com.myy.rabbitmqapi.consumer;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;import java.io.IOException;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 16:07** 具体的短信微服务收到消息的处理方案*/
public class MyConsumer extends DefaultConsumer {public MyConsumer(Channel channel) {super(channel);}@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.err.println("-----------consume message----------");System.err.println("consumerTag: " + consumerTag);System.err.println("envelope: " + envelope);System.err.println("properties: " + properties);System.err.println("body: " + new String(body));}}

消费端调用

package com.myy.rabbitmqapi.consumer;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 16:09** 短信微服务*/
public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.59.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();String exchangeName = "test_consumer_exchange";String routingKey = "consumer.#";String queueName = "test_consumer_queue";channel.exchangeDeclare(exchangeName, "topic", true, false, null);channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, exchangeName, routingKey);channel.basicConsume(queueName, true, new MyConsumer(channel));}
}

生产端调用

package com.myy.rabbitmqapi.consumer;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** @author 熊贝贝* @site www.myy.com* @company* @create  2020-02-29 16:09** 支付微服务*/
public class Producer {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.59.128");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();String exchange = "test_consumer_exchange";String routingKey = "consumer.save";String msg = "Hello RabbitMQ Consumer Message";for(int i =0; i<5; i ++){channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());}}
}

运行消费端，再运行生产端
在这里插入图片描述
接收到消息，总五条
在这里插入图片描述

相比while循环更加便捷

http://www.lbrq.cn/news/1249003.html

相关文章：

  • 如何让我的网站被百度收录/淘宝摄影培训推荐
  • 南充建网站/网站seo具体怎么做?
  • 做最好的导航网站/搜索引擎网站有哪些
  • VS2010做网站登录页面步骤/网站没有友情链接
  • 网络工程和软件工程哪个好/郑州网站优化渠道
  • 运河经济开发区建设局网站/站长工具的网址
  • 网站建设公司浙江/互联网广告营销是什么
  • 榆林做网站/防晒霜营销软文
  • 外国做问卷可以赚钱的网站/正规的推文平台
  • 网站建设业务员招聘/阿里云万网域名查询
  • 建站网站教程视频/淘宝运营培训机构
  • 海淀网站设计公司/郑州网站建设哪家好
  • 免费网站建设ppt/重庆seo技术教程博客
  • 烟台市建委网站/开发网站用什么软件
  • 网站建设说明/西安官网seo公司
  • 网站制作推广/镇江seo公司
  • 缪斯国际设计公司官网/鸡西seo
  • 大气蓝色企业网站模板/企业做推广有用吗
  • 网站建设和编程/华为seo诊断及优化分析
  • 医疗网站优化怎么做/搜索引擎优化网页
  • 大连科技网站制作/爱链接
  • 传奇页游平台/枫林seo工具
  • 网站建设受众/semi final
  • 嘉兴企业网站建设/关键词搜索排名软件
  • 哪个网站可以做化学实验/网站建设公司业务
  • 网站建设zhuitiankeji/百度大数据
  • 网站空间和数据库空间/谷歌seo是什么
  • 网站设计首页框架图片/天天外链官网
  • 用vs2010做的网站的源码/优化推广关键词
  • 济南做企业网站公司/营销型公司网站建设
  • 透明矿山：科技重塑矿业未来
  • 全栈：如何操作在SQLserver里面CRUD（增删改查）
  • jetson上使用opencv的gstreamer进行MIPI和USB摄像头的连接以及udp推流
  • 数据结构(五)：顺序循环队列与哈希表
  • ChatGPT以及ChatGPT强化学习步骤
  • promise类方法