案例学习--016 消息队列作用和意义

news2024/10/5 19:33:45

简介

MQ全称为Message Queue, 是一种分布式应用程序的的通信方法，它是消费-生产者模型的一个典型的代表，producer往消息队列中不断写入消息，而另一端consumer则可以读取或者订阅队列中的消息。主要产品有：ActiveMQ、RocketMQ、RabbitMQ、Kafka

以rabbitMQ为例，介绍一下rabbitMQ的削峰操作。要对流量进行削峰，常用解决方案就是用消息队列来缓冲瞬时流量，把同步的直接调用转换成异步的间接推送，中间通过一个队列在一端承接瞬时的流量洪峰，在另一端平滑地将消息推送出去。

队列和消息的大小限制:

消息大小不超过4MB

队列长度限制：

消息的数量，或者是消息的总字节数，可以使用policy命令或在代码中声明来设置队列的最大长度。如果超过就有溢出操作，例如从队列的头部丢弃最老的消息。

policy命令式

定义一个消息生产者

实现方案：

定义一个消息生产者

@Test
public void test() throws Exception {
    for (int i = 0; i < 1000 ; i++) {
        rabbitTemplate.convertAndSend("test-queue ",  "消息发送);
    }
    Thread.sleep(1000 * 1000);
}

使用@RabbitListener注解定义一个消息消费者

@Component
@RabbitListener(queuesToDeclare = @Queue(name = "test-queue"))
public class Consumer {
    private int count = 0;
    @RabbitHandler
    public void receive(String msg, Channel channel, Message message) throws IOException {

        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            Thread.sleep(1000);
            System.out.println("=====消息处理===>");
            channel.basicAck(deliveryTag, true);
            System.out.println("current count is:" + ++count);
        } catch (Exception e) {

        }
    }
}

采用手动应答方式，消费者将消息消费以后通知RabbitMQ消息队列本条消息已经被消费，可以删除。(自动应答不管消费者对消息处理是否成功，都会告诉队列删除消息)

如何避免消息丢失：如果消费者由于某些原因失去连接（网络阻塞或消费者挂机），导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果没有此应答机制，RabbitMQ一旦向消费者传递了一条消息，便立即将该条消息标记为删除。在这种情况下，突然有个消费者挂掉了，将会丢失正在处理的消息和后续发给已挂机的消费者的消息。

spring.application.name=springboot_rabbitmq
spring.rabbitmq.host=39.107.96.73
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.virtual-host=/
spring.rabbitmq.listener.simple.prefetch=2

运行效果：