介绍

Spring Cloud Stream 是一个用于构建消息驱动的微服务应用程序的框架。它通过为消息中间件的不同实现提供统一的编程模型，简化了开发者使用不同消息队列系统的复杂性。其中，RabbitMQ、RocketMQ 和 Kafka 都是 Spring Cloud Stream 的支持者。下面将分别介绍如何使用 Spring Cloud Stream 与这三种消息中间件进行集成。

与 RabbitMQ 集成

1. 首先，在项目的 pom.xml 文件中添加相应的依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
XML

2. 创建一个发送消息的接口：

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

public interface MessageSource {
    String OUTPUT = "output";

    @Output(OUTPUT)
    MessageChannel output();
}
Java

3. 在 Spring Boot 启动类上添加 @EnableBinding 注解，并将刚刚创建的接口作为参数传入：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;

@SpringBootApplication
@EnableBinding(MessageSource.class)
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
Java

4. 在需要发送消息的地方，通过 MessageSource 接口的 output 方法发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class MessageSender {
    @Autowired
    private MessageSource messageSource;

    public void sendMessage(String message) {
        messageSource.output().send(MessageBuilder.withPayload(message).build());
    }
}
Java

与 RocketMQ 集成

1. 首先，在项目的 pom.xml 文件中添加相应的依赖：

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

2. 创建一个发送消息的接口：

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

public interface MessageSource {
    String OUTPUT = "output";

    @Output(OUTPUT)
    MessageChannel output();
}
Java

3. 在 Spring Boot 启动类上添加 @EnableBinding 注解，并将刚刚创建的接口作为参数传入：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;

@SpringBootApplication
@EnableBinding(MessageSource.class)
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
Java

4. 在需要发送消息的地方，通过 MessageSource 接口的 output 方法发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class MessageSender {
    @Autowired
    private MessageSource messageSource;

    public void sendMessage(String message) {
        messageSource.output().send(MessageBuilder.withPayload(message).build());
    }
}
Java

与 Kafka 集成

1. 首先，在项目的 pom.xml 文件中添加相应的依赖：

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

2. 创建一个发送消息的接口：

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

public interface MessageSource {
    String OUTPUT = "output";

    @Output(OUTPUT)
    MessageChannel output();
}
Java

3. 在 Spring Boot 启动类上添加 @EnableBinding 注解，并将刚刚创建的接口作为参数传入：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;

@SpringBootApplication
@EnableBinding(MessageSource.class)
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
Java

4. 在需要发送消息的地方，通过 MessageSource 接口的 output 方法发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class MessageSender {
    @Autowired
    private MessageSource messageSource;

    public void sendMessage(String message) {
        messageSource.output().send(MessageBuilder.withPayload(message).build());
    }
}
Java