建立网站的英语,中国建设银行官网站e路护航,美食网站建设的可行性,wordpress 安装 空白页消息队列#xff08;MQ#xff09;到底能干什么#xff1f; MQ全称为Message Queue#xff0c;也就是消息队列#xff0c;是应用程序和应用程序之间的通信方法。 在微服务盛行的当下#xff0c;MQ被使用的也是越来越多#xff0c;一般常用来进行业务异步解耦、解耦微服务…消息队列MQ到底能干什么 MQ全称为Message Queue也就是消息队列是应用程序和应用程序之间的通信方法。 在微服务盛行的当下MQ被使用的也是越来越多一般常用来进行业务异步解耦、解耦微服务、流量削峰填谷、消息分发、分布式事务的数据一致性。
1、业务异步解耦 最常见的场景就是用户注册之后需要发送注册短信、邮件通知以告知相关信息。使用MQ只需要在处理完用户信息之后给MQ发送两个消息即可邮件服务、短信服务监听MQ的任务消息根据消息进行发送即可。 直销银行中台处理支付操作将支付报文发给支付中心MQ支付中心监听到请求会处理发送第三方将最终处理结果发给中台系统接收mq。 中台系统也可以轮询查询终态。
2、流量削峰填谷 MQ比较常用的一个场景一般在秒杀、搞活动中使用广泛。使用MQ可以将需要处理的消息全部放入其中系统按照最大处理能力去获取消息进行消费这样就可以将一瞬间过来的请求分散到一段时间内进行处理避免了系统的崩溃。
3、消息分发 多个系统对同一个数据感兴趣只需要监听同一类消息即可。 例如付款系统在付款成功之后正常做法是通知外围系统这个单子付款成功了或者是外围系统定时来拉取付款结果 使用MQ后付款系统可以在付款成功之后将消息放到MQ里面想知道这个结果的系统订阅这个主题的消息即可非常方便也不需要定时去拉取数据了。
4、分布式事务的数据一致性 RocketMQ事务消息发送步骤如下 1、发送方将半事务消息发送至消息队列RocketMQ。 2、消息队列RocketMQ将消息持久化成功之后向发送方返回Ack确认消息已经发送成功此时消息为半事务消息。 3、发送方开始执行本地事务逻辑。 4、发送方根据本地事务执行结果向服务端提交二次确认Commit或是Rollback服务端收到Commit状态则将半事务消息标记为可投递订阅方最终将收到该消息服务端收到Rollback状态则删除半事务消息订阅方将不会接受该消息。
事务消息回查步骤如下 1、在断网或者是应用重启的特殊情况下上述步骤4提交的二次确认最终未到达服务端经过固定时间后服务端将对该消息发起消息回查。 2、发送方收到消息回查后需要检查对应消息的本地事务执行的最终结果。 3、发送方根据检查得到的本地事务的最终状态再次提交二次确认服务端仍按照步骤4对半事务消息进行操作。
RabbitMQ入门 Exchange交换器Exchange有4种类型: direct(默认), fanout, topic,和headers,用来接收生产者发送的消息并将这些消息路由给服务器中的队列 Queue消息队列用来保存消息直到发送给消费者。 Binding用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则所以可以将交换器理解成一个由绑定构成的路由表。
1、基本消息类型 rabbitTemplate.converAndSend(queue, message) RabbitListener(queue)
2、工作队列 rabbitTemplate.converAndSend(queue, message) 多个监听器消费
3、发布和订阅模型 3.1、Fanout广播将收到的消息路由给所有绑定的queue 定义配置类配置FanoutExchange、Queue、Binding rabbitTemplate.converAndSend(Exchange, “”, message) RabbitListener(queue)
3.2、DriectExchage将收到的消息发给指定queue 定义配置类配置DirectExchange、Queue、Binding(with指定 routingKey ) rabbitTemplate.converAndSend(Exchange, routingKey, message) RabbitListener(queues)
3.3、TopicExchange话题广播 定义配置类配置DirectExchange、Queue、Binding(with指定 routingKey ) #0或者多个 *一个单词 rabbitTemplate.converAndSend(Exchange, routingKey, message) RabbitListener(queues)
什么是事务 数据库事务。原子性、一致性、隔离性、持久性。
本地事务 整个服务操作只能涉及一个单一的数据库资源。
分布式事务 分布式事务指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。分布式事务就是为了保证不同数据库的数据一致性。
分布式事务应用架构 微服务架构的分布式应用环境下越来越多的应用要求对多个数据库资源多个服务的访问都能纳入到同一个事务当中分布式事务应运而生。
分布式事务有几种解决方案 两阶段提交/XA MQ 第一阶段prepare即所有的参与者RM准备执行事务并锁住需要的资源。参与者ready时向TM报告已准备就绪。 第二阶段 (commit/rollback)当事务管理者™确认所有参与者(RM)都ready后向所有参与者发送commit命令。
https://blog.csdn.net/qq_43410878/article/details/123656765 RabbitMQ使用详解
基本用法
1、简单队列模式
生产者 // 1、获取到通道 Channel channel RabbitMqUtils.getChannel();
// 2、声明创建队列 // queue队列的名称 // durable是否持久化 // exclusive设置是否排他。 // autoDelete设置是否自动删除。至少有一个消费者连接到这个队列之后所有与这个队列连接的消费者都断开时才会自动删除。 // arguments设置队列的其他一些参数 channel.queueDeclare(“test”, true, false, false, null);
// exchange:要将消息发送到的Exchange(交换器) // routingKey: 路由KEY // props: 消息的其它属性如路由头等 // body: 消息体 channel.basicPublish(“”, “test”, null, (i “hello”).getBytes());
消费者 // 1、获取通道 Channel channel RabbitMqUtils.getChannel();
// 2、接收到消息回调此函数 // queue:队列名 // autoAcktrue 接收到传递消息自动应答false 接收到消息后不自动应答服务器 // deliverCallback 当一个消息发送过来后的回调接口 channel.basicConsume(“test”, true, new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(Consumer-one: 收到的消息: new String(body)); } });
2、消息自动确认 channel.basicQos(1); // 批量确认模式通过设置channel.basicQos方法的prefetchCount参数为大于1的值同样需要手动调用channel.basicAck方法来确认消息。 channel.queueDeclare(“firstQueue”, false, false, false, null);
// 消息改为手动确认 channel.basicConsume(“firstQueue”, false, new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try { TimeUnit.SECONDS.sleep(4); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(new String(body, Charset.defaultCharset())); // 手动确认消息已经被消费了, 第一个参数是当前消费的消息的标签(递增的整数) // 第二个参数是否确认多条消息包括之前消费的消息 channel.basicAck(envelope.getDeliveryTag(), false); } });
发布订阅模型(fanout) exchange类型有如下3种 fanout 广播 把消费转发给所有 绑定到该交换机的所有队列 direct 定向 把消息转发给符合 指定 routing key 路由键的队列 topic 通配符 把消息交给 routing pattern路由模式的队列
exchange 交换机 只负责转发消息 不具备存储消息的能力 因此如果没有任何队列与 exchange 绑定 或者没有符合规则的队列 那么消息会丢失
每个消费者监听自己的队列 生产者把消息发送给 broker 由交换机把消息转发到绑定此交换机的所有队列交换机需要与队列绑定 绑定之后一个消息可以被多个消费者收到
// 声明一个交换机, 交换机的类型为 fanout 参数1-交换机名称 参数2-交换机类型fanout, topic, direct, headers channel.exchangeDeclare(“multiple”, BuiltinExchangeType.FANOUT);
channel.basicPublish(“multiple”, “”, null, “fanout-message”.getBytes());
RabbitMQUtils.close(channel, connection);
//创建了一个非持久的、排他的、自动删除的队列并生成了一个名称 String qName channel.queueDeclare().getQueue(); //创建临时队列 //交换机与队列进行绑定-如果没有队列与交换机进行绑定那么消费者接受不到生产者的消息消息会丢失 channel.queueBind(qName, “multiple”, “”);
channel.basicConsume(qName, true, new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, Charset.defaultCharset())); } });
直连模型(direct) 消息的发送方在 向 Exchange发送消息时也必须指定消息的 RoutingKey。 队列与交换机的绑定不能是任意绑定了而是要指定一个RoutingKey路由key
channel.exchangeDeclare(“direct-module”, BuiltinExchangeType.DIRECT);
channel.basicPublish(“direct-module”, “success”, null, “成功信息”.getBytes());
String qName channel.queueDeclare().getQueue(); //创建临时队列 channel.queueBind(qName, “direct-module”, “success”);
channel.basicConsume(qName, true, new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, Charset.defaultCharset())); } });
主题模式(topic) Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符多个单词之间以”.”分割
统配符 * (star) can substitute for exactly one word. 匹配不多不少恰好1个词# (hash) can substitute for zero or more words. 匹配一个或多个词如: audit.# 匹配audit.irs.corporate或者audit.irs 等audit.* 只能匹配 audit.irs// 声明一个交换机, 交换机的类型为 fanout channel.exchangeDeclare(“topic-module”, BuiltinExchangeType.TOPIC);
channel.basicPublish(“topic-module”, “company.java”, null, “通知信息”.getBytes());
RabbitMQUtils.close(channel, connection);
String qName channel.queueDeclare().getQueue(); //创建临时队列 channel.queueBind(qName, “topic-module”, “company.#”); channel.queueBind(qName, “topic-module”, “company.java.#”); channel.queueBind(qName, “topic-module”, “company.html.*”);
channel.basicConsume(qName, true, new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, Charset.defaultCharset())); } });
RabbitMQ与springboot整合 1、添加依赖 spring-boot-starter-amqp 2、配置RabbitConfig包含RabbitListenerContainerFactory、RabbitTemplate
简单模型 SpringBootTest class SpringbootRabbitmqApplicationTests { Resource private RabbitTemplate rabbitTemplate;
Test
public void simpleMessageSend() {rabbitTemplate.convertAndSend(simpleQueue, new User(1, 张));
}}
RabbitListener(queuesToDeclare {Queue(“simpleQueue”)}) public void simpleModel(User user) { log.info(“message: {}”, user); }
发布订阅模型 // fanout模型 Test public void fanoutMessageSend() { for (int i 0; i 5; i) { rabbitTemplate.convertAndSend(“fanout-ex”, “”, new User(i, “张三”)); } try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); } }
// valueQueue 创建临时队列 // exchange创建交换机 RabbitListener(bindings { QueueBinding(value Queue, exchange Exchange(value “fanout-ex”, type ExchangeTypes.FANOUT)) }) public void receiveMessage1(User user) { System.out.println(String.format(“消费者 【one】: %s”, user)); }
RabbitListener(bindings { QueueBinding(value Queue, exchange Exchange(value “fanout-ex”, type ExchangeTypes.FANOUT)) }) public void receiveMessage2(User user) { System.out.println(String.format(“消费者 【two】: %s”, user)); }
直连模式(direct) Test public void directMessageSend() { //rabbitTemplate.convertAndSend(“direct-ex”, “success”, new User(2, “张三”)); rabbitTemplate.convertAndSend(“direct-ex”, “error”, new User(2, “张三”)); try { TimeUnit.SECONDS.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } }
RabbitListener(bindings { QueueBinding(value Queue, key {“error”, “success”}, exchange Exchange(value “direct-ex”, type ExchangeTypes.DIRECT)) }) public void receiveMessage1(User user) { System.out.println(String.format(“消费者 【one】: %s”, user)); }
RabbitListener(bindings { QueueBinding(value Queue, key {“error”}, exchange Exchange(value “direct-ex”, type ExchangeTypes.DIRECT)) }) public void receiveMessage2(User user) { System.out.println(String.format(“消费者 【two】: %s”, user)); }
topic模型 Test public void topicMessageSend() { //rabbitTemplate.convertAndSend(“topic-ex”, “company”, new User(2, “张三”)); rabbitTemplate.convertAndSend(“topic-ex”, “company.java”, new User(2, “张三”)); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } }
RabbitListener(bindings { QueueBinding(value Queue, key {“company.#”}, exchange Exchange(value “topic-ex”, type ExchangeTypes.TOPIC)) }) public void receiveMessage1(User user) { System.out.println(String.format(“消费者 【one】: %s”, user)); }
RabbitListener(bindings { QueueBinding(value Queue, key {“company.java.#”}, exchange Exchange(value “topic-ex”, type ExchangeTypes.TOPIC)) }) public void receiveMessage2(User user) { System.out.println(String.format(“消费者 【two】: %s”, user)); }
RabbitListener(bindings { QueueBinding(value Queue, key {“company.html.*”}, exchange Exchange(value “topic-ex”, type ExchangeTypes.TOPIC)) }) public void receiveMessage3(User user) { System.out.println(String.format(“消费者 【three】: %s”, user)); }
消息的手动确认 spring: rabbitmq: listener: simple: # 提交方式为手动 acknowledge-mode: MANUAL
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
面试a. 如何保证消息不丢失
b. 如何保证消息的不重复消费
c. 如何使用mq来是实现分布式事务
d. 在工作中mq用在哪里支付回调。
RabbitMQ如何做到消息不丢失
1.持久化
发送消息时设置delivery_mode属性为2使消息被持久化保存到磁盘即使RabbitMQ服务器宕机也能保证消息不丢失。同时创建队列时设置durable属性为True以确保队列也被持久化保存。
2.确认机制
消费者通过basic.ack命令向RabbitMQ服务器确认已经消费了消息。如果消费者处理消息时发生错误或宕机RabbitMQ会重新将消息发送给其他消费者。RabbitMQ在接收到消费者确认消息前会将消息保存在内存中在确认后才会删除消息。
3.发布者确认
RabbitMQ支持发布者确认机制即发布者在将消息发送到队列后等待RabbitMQ服务器的确认消息。成功保存到队列的消息会返回确认消息给发布者如果无法保存则返回NackNegative Acknowledgement消息。通过发布者确认机制可以确保消息成功发送到RabbitMQ服务器。
4.备份队列
RabbitMQ支持备份队列Alternate Exchange机制即在消息发送到队列之前先将消息发送到备份队列。如果主队列无法接收消息RabbitMQ会将消息发送到备份队列中。备份队列通常是一个交换机在创建队列时可以通过x-dead-letter-exchange属性指定备份队列。
三、注意事项 在使用RabbitMQ消息确认机制时需注意以下几点
1、确认模式的选择 根据具体业务需求选择合适的确认模式。如果对消息传递的可靠性要求较高建议使用手动确认模式或批量确认模式。
2、设置消息持久化 在发布消息时通过设置消息的持久化属性可以确保即使RabbitMQ服务器意外关闭或重启消息仍然能够被保存下来。
3、处理未确认消息 如果消费者在处理消息时发生异常导致无法发送确认信号给RabbitMQ那么这些消息将保持在队列中并可能被重新投递。需要设定适当的重试策略和错误处理机制来处理这些未确认的消息。
4、监听确认超时 在使用批量确认模式时如果长时间没有收到确认信号需要设置合理的超时时间并采取相应措施来处理超时的情况。
RabbitMQ消息确认机制详解确保交付成功
消息确认机制是通过发布者Producer和消费者Consumer之间的交互来实现的。 当发布者发送消息到RabbitMQ后会等待确认结果。如果消息成功被消费者接收并处理消费者会发送一个确认信号给RabbitMQ告知消息已经处理完成。而RabbitMQ则会根据接收到的确认信号判断消息是否成功交付。 消息确认机制一般有两种模式简单模式和批量模式。在简单模式中每发送一条消息后就等待确认而在批量模式中可以一次发送多条消息然后等待批量确认。无论是简单模式还是批量模式都可以提高消息传递的可靠性。
https://blog.csdn.net/weixin_52278591/article/details/128841155 RabbitMq 消息确认机制详解
https://blog.csdn.net/u011942456/article/details/128198956 总结RabbitMq消息丢失和消息重复消费问题
https://www.zhihu.com/question/54756562?utm_id0 rabbitmq是什么主要用于哪些方面
https://zhuanlan.zhihu.com/p/583520436?utm_id0 用了8年MQ聊聊消息队列的技术选型哪个最香
https://blog.csdn.net/qq_43410878/article/details/123656765 RabbitMQ使用详解
https://zhuanlan.zhihu.com/p/590948541 绝对详细的 RabbitMQ入门看完本系列就够了
