SpringCloudStream整合RabbitMQ和Kafka

Spring Cloud Stream 是基于 Spring Boot 和 Spring Integration 的框架，用于构建消息驱动的微服务应用程序。它提供了一种简单的方式来在应用程序中使用消息中间件，比如 RabbitMQ 和 Kafka。

下面分别介绍 Spring Cloud Stream 如何整合 RabbitMQ 和 Kafka。

## 整合 RabbitMQ

1. 添加依赖

在 `pom.xml` 文件中添加以下依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
</dependency>

2. 创建消息生产者

创建一个消息生产者，可以使用 `@EnableBinding` 注解将其与 RabbitMQ 绑定：

@EnableBinding(MessageSource.class)
public class RabbitMQProducer {
    @Autowired
    private MessageSource<String> messageSource;
    
    public void sendMessage(String message) {
        messageSource.output().send(MessageBuilder.withPayload(message).build());
    }
}

3. 创建消息消费者

创建一个消息消费者，同样使用 `@EnableBinding` 注解将其与 RabbitMQ 绑定：

@EnableBinding(MessageSink.class)
public class RabbitMQConsumer {
    @StreamListener(MessageSink.INPUT)
    public void receiveMessage(String message) {
        System.out.println("Received: " + message);
    }
}

4. 配置 RabbitMQ

在 `application.yml` 文件中配置 RabbitMQ：

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

## 整合 Kafka

1. 添加依赖

在 `pom.xml` 文件中添加以下依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

2. 创建消息生产者

创建一个消息生产者，可以使用 `@EnableBinding` 注解将其与 Kafka 绑定：

@EnableBinding(MessageSource.class)
public class KafkaProducer {
    @Autowired
    private MessageSource<String> messageSource;
    
    public void sendMessage(String message) {
        messageSource.output().send(MessageBuilder.withPayload(message).build());
    }
}

3. 创建消息消费者

创建一个消息消费者，同样使用 `@EnableBinding` 注解将其与 Kafka 绑定：

@EnableBinding(MessageSink.class)
public class KafkaConsumer {
    @StreamListener(MessageSink.INPUT)
    public void receiveMessage(String message) {
        System.out.println("Received: " + message);
    }
}

4. 配置 Kafka

在 `application.yml` 文件中配置 Kafka：

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