消息队列的原理与基本使用

消息队列

定义

消息队列（message queue）是一种用于在软件系统中传输、存储和处理消息的机制。它通常用于异步通信，允许不同的组件或系统在时间上解耦，以便它们可以独立地生产和消费消息。消息队列通常由一个服务或软件组件管理，它负责接收、存储和传递消息，并确保消息按照一定的顺序和规则进行处理。

消息队列的基本原理是将消息从一个发送者发送到一个或多个接收者。发送者将消息放入队列中，而接收者则从队列中获取消息进行处理。

优点 与 缺点

优点

• 解耦性：生产者者和消费者之间不直接通信，而是通过队列进行消息传递，从而减少了组件之间的耦合度。如果某个服务挂掉，不会影响其他服务，可以通过后期进行重新补偿。
• 异步性：生产者可以继续执行而无需等待接收者的响应，提高了系统的并发性和性能。每次多增加服务，只需要多添加一个消费者即可。
• 削峰性：队列可以暂时存储消息，以防止消息丢失或超载。

其实就是相当于一个暂存箱，存入数据，然后不同功能对其进行异步处理，而不像传统的同步处理。这样就很好的解释了解耦和异步的特点。

而对于削峰，也很好理解，比如在购物秒杀系统中，某一瞬间，会突然来大量的订单，如果采用传统的方式，会导致服务器负载过高，甚至崩溃，使用消息队列可以作为缓冲，将订单先暂存下来，等待后端后面慢慢的处理，从而更好的应对高峰流量。

这样就实现了系统的高可用、高性能、高并发。

缺点

• 复杂性增加: 引入消息队列会增加系统的复杂度，需要额外的部署、维护和监控。

• 消息丢失: 在一些情况下，消息队列可能会出现消息丢失的情况，例如由于网络问题或者队列本身的问题导致消息无法正确传递。需要对消息进行适当的备份。

• 一致性问题（事务问题）: 如果消息队列中的消息被重复消费或者消费顺序不正确，可能会导致系统出现一致性问题。同时也需要考虑服务的事务性，考虑补偿处理或事务处理。

• 性能损耗: 引入消息队列会增加系统的延迟和资源消耗，尤其是在消息队列的读写操作上。

常见的消息队列

RabbitMQ：RabbitMQ 是一个开源的消息队列系统，基于 AMQP（高级消息队列协议）实现。它支持多种消息传输模式，包括点对点、发布/订阅、路由等，具有高度的灵活性和可定制性。RabbitMQ 提供了丰富的功能，包括消息持久化、消息确认、消息队列监控等，被广泛应用于分布式系统和微服务架构中。

Apache Kafka：Apache Kafka 是一个分布式流处理平台和消息队列系统，最初由 LinkedIn 开发并开源。它设计用于处理大规模的实时数据流，具有高吞吐量、持久性、可伸缩性和容错性。Kafka 采用分布式的日志架构，消息被持久化存储在磁盘上，并通过多个分区进行分布式处理，常用于日志收集、事件驱动架构等场景。

Amazon SQS：Amazon Simple Queue Service（SQS）是亚马逊提供的一种完全托管的消息队列服务。它具有高度可靠性、可伸缩性和安全性，可以在不同的应用程序之间可靠地传输消息。SQS 支持标准队列和 FIFO 队列两种类型，适用于不同的使用场景。

Apache ActiveMQ：Apache ActiveMQ 是一个基于 Java 实现的开源消息队列系统，支持多种通信协议，包括 OpenWire、STOMP、AMQP、MQTT 等。它具有丰富的特性，包括消息持久化、事务支持、集群和负载均衡等，被广泛应用于企业级应用和集成系统中。

Redis：Redis 是一个开源的内存数据库和缓存系统，同时也可以用作消息队列。Redis 的发布/订阅功能可以实现简单的消息传递，而 Redis 的列表数据结构也可以用于构建基本的队列系统。尽管 Redis 不是专门设计用作消息队列，但在一些场景下可以作为简单的消息传递机制使用。

消息队列的配置

这里使用的Kafka
导入依赖：

        
        <dependency>
            <groupId>org.springframework.kafkagroupId>
            <artifactId>spring-kafkaartifactId>
        dependency>
1
2
3
4
5

配置参数:

spring:
  redis:
    host: 填写ip
    port: 6379
    password: 123456
  kafka:
    bootstrap-servers: 填写ip:9092
    consumer:
      group-id: 
1
2
3
4
5
6
7
8
9

• bootstrap-servers：指定 Kafka 集群中的一个或多个 broker 地址。填写 Kafka 集群中任意一个 broker 的 IP 地址和端口号。客户端通过连接这些 broker 来发现整个 - - Kafka 集群的拓扑结构，并进行数据交换。
• consumer：这一部分可能是消费者配置的一部分，指定了消费者的一些参数。
• group-id：指定消费者所属的消费者组 ID。消费者组是 Kafka 中一个重要的概念，它允许多个消费者协同工作来处理主题中的消息。同一个消费者组内的消费者共享消息，但每个消息只能被消费者组内的一个消费者处理。这个配置项定义了消费者所属的组 ID。

消息队列的简单使用

生产者

编写controller，和其他一样，注入KafkaTemplate使用即可。

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;

import javax.annotation.Resource;

@RestController
public class KafkaController {

    @Resource
    private KafkaTemplate kafkaTemplate;

    @GetMapping("/produce")
    public String produce(String message) {
        kafkaTemplate.send("order", message);

        return "发消息成功";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

• send方法，发送消息到order队列中。

消费者

编写listen，监听消息队列中的消息。

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaListen {

    @KafkaListener(topics = {"order"})
    public void listen(ConsumerRecord<String, String> record) {

        System.out.println("卡夫卡推送的消息内容：" + record.value());
        System.out.println("我收到了消息");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

• @KafkaListener(topics = {“order”}) 表示监听order这个消息队列
• ConsumerRecord record 表示要处理的数据

运行

例如我请求produce发送三条消息，1，2，3到消息队列
消费者就能监听到