MQ简介

什么是MQ跨进程消息队列？它的主要角色包括生产者和消费者。生产者只负责生产信息；他们不知道消费者是谁、他们将如何处理消息或结果是什么。消费者负责接收和处理消息，但无法知道生产者是谁或消息是如何产生的。 MQ能做什么？ MQ 的特性通常包括异步、高吞吐量和低延迟，使其适用于：限制电流并削减高峰和低谷。可靠事件、数据一致性处理。有几个功能允许您运行计划任务。等等. MQ处理消息是异步的，所以如果需要及时返回处理结果，就不要使用MQ。

MQ系统有什么缺点：除了代码组件的访问之外，还必须考虑高可用性、集群、消息可靠性等问题。 Producer：如何保证消息发送可靠，如何避免消息重复？ Consumer：如何保证幂等性，处理重复消息也会出现处理延迟等问题。 优点：解耦，MQ可以让你很好的解耦系统，尤其是分布式/微服务系统。

原本同步的操作可以异步处理，从而获得更快的响应时间。

可以使用MQ的场景（一）

用户系统或其他需要发送短信的系统的系统隔离可以通过MQ来完成。用户系统和文本消息系统是完全分开的。

场景（2）

顺序执行任务场景。假设三个任务A B C，其中B必须等待A完成才能执行，C必须等待B完成才能执行。

同样，假设定时器A 在9 点钟运行，定时器B 在10 点钟运行，定时器C 在11 点钟运行。

这实际上是不安全的，因为后一个任务无法知道前一个任务是否真正被执行。假设A宕机了，B在10点定时运行。这时候数据就变得异常了。

通过MQ，A 在执行完成后向B 发送消息。 C收到B的消息后同样执行。

场景(3)

支付网关通知通常要求我们的系统访问支付功能。通常，支付宝以回调的形式通知系统支付结果。

支付网关可以通过MQ解耦并与业务系统通信进行处理，有助于系统的解耦和扩展。

假设当用户成功支付时，积分就会被添加到用户身上。您所需要的只是一个积分系统来监控此消息并处理积分。无需更改或修改网关层代码。

即使我不使用MQ，我是否需要更改我的网关系统上的代码并远程调用接口以获得更多积分？

这就是使用MQ解耦和扩展的好处。

当然，转发规则还必须保证涉及到的每个队列都能收到消息。

场景(4)

微服务/分布式系统、分布式事务——最终一致的处理解决方案！

详细信息：分布式事务处理解决方案、微服务事务处理解决方案

场景(5)

一个消息延迟队列可以运行多个定时任务，即不固定时间运行的定时任务。示例：如果用户下单后24小时内未付款，订单将被取消。如果确认收货后2天内没有评分，则会自动评分。通常的方法是启用计时器并在每次需要处理订单或其他数据时运行外卖。

这种做法的一个原因是，如果数据量很大，每次都必须扫描数据库，效率很低。

而且也不是很及时，最坏的情况下可能要等一轮。

执行结果也可能重复，导致难以平衡时效性和轮询频率。

利用MQ (Rabbitmq)、DLX（死信交换）和消息TTL（生存时间扩展）功能。您可以高效地完成此任务场景。无需扫描图书馆，提高时效性。

DLX：http://www.rabbitmq.com/dlx.html，

TTL：http://www.rabbitmq.com/ttl.html#per-message-ttl

原则：

您可以为发送到队列的消息设置生存时间TTL。如果消息在其生命周期内未被消耗，则可以将其传输到另一个队列并配置为运行另一个队列中的任务。这意味着可以达到消息延迟的效果。

设置到期日期：

可以为消息队列统一设置过期日期，也可以为特定消息单独设置过期日期。

如果PS消息有过期时间，到达有过期时间的队列时，优先以该队列设置的过期时间为准。

以SpringBoot为例。

配置传输队列和传输队列。

@Component@Configurationpublic class RabbitMqConfig { @Bean public Queue curQueue() { MapString, Object args=new HashMapString, Object() //超时后转发器转发到lay_queue_exchange args.put('x-dead-letter-exchange ', 'delay_queue_exchange '); //routingKey转发规则args.put('x-dead-letter-routing-key', 'user.#') //过期时间20秒args.put('x-message- ttl ', 20000) ; return new Queue('cur_queue', false, false, false, args); } @Bean public Queue LateQueue() { return new Queue('delay_queue') } @Bean TopicExchangeexchange(); cur_queue_exchange'); } @Bean TopicExchangeexchange2() { //转发队列return new TopicExchange('delay_queue_exchange') } @Bean Binding bindingHelloQueue(Queue curQueue, TopicExchangeexchange) { //队列到绑定转发器return BindingBuilder.bind(curQueue).to (exchange).with('user.#'); } @Bean Binding bindingHelloQueue2(Queue LateQueue, TopicExchangeexchange2) { return BindingBuilder.bind(layQueue).to (Exchange2).with('user .#');显示：

@Componentpublic class MqEventSender { Logger logger=LoggerFactory.getLogger(MqEventSender.class); @Autowired private RabbitTemplate RabbitTemplate; /** * 消息没有设定时间，排队cur_queue_exchange * @param msg */public void sendMsg( String msg) { logger.info('发送消息： ' + msg); String msg) { logger.info('发送消息： ' + MessageProperties messageProperties=new MessageProperties(); //设置过期时间为10 秒messageProperties.setExpiration('10000');toMessage(msg, messageProperties); RabbitTemplate.convertAndSend('cur_queue_exchange', 'user.ss', message);

监听队列不是cur_queue，而是lay_queue。

PS cur_queue 不应包含侦听器。如果没有监听器，消息将无法被消费以达到想要的延迟消息效果。

/** * 由linli于2017/8/21创建。 * 监听扔进超时队列的内容*/@Component@RabbitListener(queues='delay_queue') public class DelayQueueListener { public static Logger logger=LoggerFactory.getLogger( AddCommentsEventListener.class); ) { logger.info('收到消息'+msg);

/** * 由linli于2017/8/21创建。 */@RestController@RequestMapping('/test') public class TestContorller { @Autowired MqEventSender sender; @RequestMapping('/mq/delay') public String test() { sender.sendMsg('将延迟消息添加到队列！' ) ; sender.sendMsgWithTime('消息延迟！');

根据我们的观察，发送和接收时间的间隔为20秒。

您为消息设置的10 秒TTL 时间未生效。 Validated：如果消息有过期时间，并且到达有过期时间的队列，则队列设置的过期时间优先。

如果您希望每条消息都有自己的生存时间并发送到队列，请不要设置它。

args.put(“x-message-ttl”, 20000);

是否对队列或消息设置消息过期取决于您独特的业务场景。

总结MQ 是一个进程间消息队列，可以有效地用于隔离系统。

MQ的引入给系统增加了一些复杂性，必须进行评估。

MQ 适用于异步任务，但不适用于同步任务。

版权声明：本文转载于今日头条，版权归作者所有。如有侵权，请联系本站编辑删除。