35. 如何在Spring Boot中配置消息队列（如RabbitMQ、Kafka）？

在 Spring Boot 中配置消息队列（如 RabbitMQ 和 Kafka）非常简单。Spring Boot 提供了与这些消息队列集成的起步依赖（starters），并且通过自动配置和 Spring 的强大支持，可以快速地实现消息队列的生产和消费功能。

1. 使用 RabbitMQ

1.1 引入 RabbitMQ 依赖

首先，在 Spring Boot 项目中引入 spring-boot-starter-amqp 依赖，这个依赖包含了与 RabbitMQ 的集成支持。

Maven 配置：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Gradle 配置：

implementation 'org.springframework.boot:spring-boot-starter-amqp'

1.2 配置 RabbitMQ

在 application.properties 或 application.yml 中配置 RabbitMQ 的连接信息。

application.properties 示例：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

application.yml 示例：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

1.3 创建消息队列的配置类

你可以通过一个配置类来定义队列、交换机、绑定等 RabbitMQ 的核心组件。

示例：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    public static final String QUEUE_NAME = "myQueue";
    public static final String EXCHANGE_NAME = "myExchange";
    public static final String ROUTING_KEY = "myRoutingKey";

    @Bean
    public Queue myQueue() {
        return new Queue(QUEUE_NAME, true);
    }

    @Bean
    public TopicExchange myExchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public Binding binding(Queue myQueue, TopicExchange myExchange) {
        return BindingBuilder.bind(myQueue).to(myExchange).with(ROUTING_KEY);
    }
}

1.4 生产和消费消息

生产消息：

你可以创建一个服务类，通过 AmqpTemplate 来发送消息。

示例：

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class RabbitMQProducer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendMessage(String message) {
        amqpTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, message);
        System.out.println("Message sent: " + message);
    }
}

消费消息：

你可以通过 @RabbitListener 注解来监听队列，并消费消息。

示例：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class RabbitMQConsumer {

    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

2. 使用 Kafka

2.1 引入 Kafka 依赖

在 Spring Boot 项目中引入 spring-boot-starter-kafka 依赖，这个依赖包含了与 Kafka 的集成支持。

Maven 配置：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-kafka</artifactId>
</dependency>

Gradle 配置：

implementation 'org.springframework.boot:spring-boot-starter-kafka'

2.2 配置 Kafka

在 application.properties 或 application.yml 中配置 Kafka 的连接信息。

application.properties 示例：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup
spring.kafka.consumer.auto-offset-reset=earliest

application.yml 示例：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: myGroup
      auto-offset-reset: earliest

2.3 创建 Kafka 的配置类

如果你需要进一步自定义 Kafka 的配置，可以创建一个配置类。

示例：

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaConfig {

    @Bean
    public NewTopic myTopic() {
        return new NewTopic("myTopic", 1, (short) 1);
    }
}

2.4 生产和消费消息

生产消息：

创建一个服务类，通过 KafkaTemplate 来发送消息。

示例：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String message) {
        kafkaTemplate.send("myTopic", message);
        System.out.println("Message sent: " + message);
    }
}

消费消息：

通过 @KafkaListener 注解来监听主题，并消费消息。

示例：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumer {

    @KafkaListener(topics = "myTopic", groupId = "myGroup")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}

3. 总结

• RabbitMQ 集成：通过引入 spring-boot-starter-amqp 依赖，配置 application.properties，并使用 AmqpTemplate 和 @RabbitListener 实现消息的生产和消费。
• Kafka 集成：通过引入 spring-boot-starter-kafka 依赖，配置 application.properties，并使用 KafkaTemplate 和 @KafkaListener 实现消息的生产和消费。

Spring Boot 提供的 starters 让消息队列的集成变得非常简便，开发者可以快速地在项目中引入 RabbitMQ 或 Kafka，实现可靠的消息传输和处理。