怎么使用Spring Cloud Stream玩转RabbitMQ，RocketMQ和Kafka

介绍

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>

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

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

public interface MessageSource {
    String OUTPUT = "output";

    @Output(OUTPUT)
    MessageChannel output();
}

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);
    }
}

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());
    }
}

与RocketMQ集成

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

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

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

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

public interface MessageSource {
    String OUTPUT = "output";

    @Output(OUTPUT)
    MessageChannel output();
}

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);
    }
}

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());
    }
}

与Kafka集成

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

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

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

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

public interface MessageSource {
    String OUTPUT = "output";

    @Output(OUTPUT)
    MessageChannel output();
}

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);
    }
}

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());
    }
}