Spring Cloud Stream 简介

Spring Cloud Stream 是一个框架，用于构建与共享消息系统连接的高度可扩展的事件驱动微服务。 该框架提供了一个灵活的编程模型，该模型基于已经建立和熟悉的 Spring 习惯用法和最佳实践，包括对持久 pub/sub 语义、消费者组和有状态分区的支持。

Spring Cloud Stream支持哪些消息中间件

Spring Cloud Stream 支持的消息中间件如下图所示：

快速使用

设置依赖（pom）

对于依赖，我们需要注意 spring-boot``spring-cloud``spring-cloud-alibaba版本对应关系即可。

    <properties>        <maven.compiler.target>1.8</maven.compiler.target>        <maven.compiler.source>1.8</maven.compiler.source>        <spring.boot.version>2.2.4.RELEASE</spring.boot.version>        <spring.cloud.version>Hoxton.SR1</spring.cloud.version>        <spring.cloud.alibaba.version>2.2.0.RELEASE</spring.cloud.alibaba.version>    </properties>    <dependencyManagement>        <dependencies>            <dependency>                <groupId>org.springframework.boot</groupId>                <artifactId>spring-boot-starter-parent</artifactId>                <version>${spring.boot.version}</version>                <type>pom</type>                <scope>import</scope>            </dependency>            <dependency>                <groupId>org.springframework.cloud</groupId>                <artifactId>spring-cloud-dependencies</artifactId>                <version>${spring.cloud.version}</version>                <type>pom</type>                <scope>import</scope>            </dependency>            <dependency>                <groupId>com.alibaba.cloud</groupId>                <artifactId>spring-cloud-alibaba-dependencies</artifactId>                <version>${spring.cloud.alibaba.version}</version>                <type>pom</type>                <scope>import</scope>            </dependency>        </dependencies>    </dependencyManagement>    <dependencies>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>        </dependency>        <dependency>            <groupId>com.alibaba.cloud</groupId>            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>        </dependency>    </dependencies>

设置配置（yml）

配置  rocketmq需要的一些基础配置信息。

spring:  cloud:    stream:      bindings:        input:          destination: TestTopic          group: TestGroup        output:          destination: TestTopic    rocketmq:      binder:        name-server: 127.0.0.1:9876server:  port: 18080

设置启动类

@EnableBinding({Source.class, Sink.class})@SpringBootApplicationpublic class RocketMQApplication {    public static void main(String[] args) {        SpringApplication.run(RocketMQApplication.class, args);    }}

消息生产者

主要是实现发消息

@Componentpublic class RocketMQProducer {    @Resource    private Source source;    public void sendMessage(String msg) {        Map<String, Object> headers = new HashMap<>();        headers.put(MessageConst.PROPERTY_TAGS, "testTag");        MessageHeaders messageHeaders = new MessageHeaders(headers);        Message<String> message = MessageBuilder.createMessage(msg,                messageHeaders);        this.source.output().send(message);    }}

消息消费者

主要是用来接受消息， 并且打印消息的内容。

@Componentpublic class TestTopicConsumer {    @StreamListener(Sink.INPUT)    public void onMessage(String messsage) {        System.out.println("received message:" + messsage + " from binding:" + Sink.INPUT);    }}

测试

我们编写一个 TestController类，对外提供一个发消息的接口，去调用发消息，实现发消息。

@RestController@RequestMapping("/MQTest")public class TestController {    @Resource    private RocketMQProducer producer;    @RequestMapping("/sendMessage")    public String sendMessage(String message) {        producer.sendMessage(message);        return "send message success";    }}

测试发消息，执行命令 curl http://127.0.0.1:18080/MQTest/sendMessage\?message\=hahha , 返回结果如下： 服务端控制台输出如下：

总结

本文简答的对 Spring Cloud Stream  整合 RockectMQ 做了实践，但是在我们在实际开发过程中，这样来做是远远不够的，注意这个只是案例，后面我也会分享项目中正真使用的方案。

原文：https://juejin.cn/post/7104282113075068965