Spring cloud stream【入门介绍】

案例代码:https://github.com/q279583842q/springcloud-e-book

??在实际开发过程中，服务与服务之间通信经常会使用到消息中间件，而以往使用了哪个中间件比如RabbitMQ，那么该中间件和系统的耦合性就会非常高，如果我们要替换为Kafka那么变动会比较大，这时我们可以使用SpringCloudStream来整合我们的消息中间件，来降低系统和中间件的耦合性。

一、什么是SpringCloudStream

??官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。
??应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream 中binder 交互，通过我们配置来 binding ，而 Spring Cloud Stream 的 binder 负责与消息中间件交互。所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。
??通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。目前仅支持RabbitMQ、Kafka。

二、Stream 解决了什么问题?

??Stream解决了开发人员无感知的使用消息中间件的问题，因为Stream对消息中间件的进一步封装，可以做到代码层面对中间件的无感知，甚至于动态的切换中间件(rabbitmq切换为kafka)，使得微服务开发的高度解耦，服务可以关注更多自己的业务流程

官网结构图

组成 说明
Middleware 中间件，目前只支持RabbitMQ和Kafka
Binder Binder是应用与消息中间件之间的封装，目前实行了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic，RabbitMQ的exchange)，这些都可以通过配置文件来实现
@Input 注解标识输入通道，通过该输入通道接收到的消息进入应用程序
@Output 注解标识输出通道，发布的消息将通过该通道离开应用程序
@StreamListener 监听队列，用于消费者的队列的消息接收
@EnableBinding 指信道channel和exchange绑定在一起

三、消息驱动入门案例

??我们通过一个入门案例来演示下通过stream来整合RabbitMQ来实现消息的异步通信的效果，所以首先要开启RabbitMQ服务，RabbitMQ不清楚的请参考此文:https://dpb-bobokaoya-sm.blog.csdn.net/article/details/90409404

1.创建消息发送者服务

1.1 创建项目

??创建一个SpringCloud项目

1.2 pom文件

??pom文件中重点是要添加spring-cloud-starter-stream-rabbit这个依赖

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.13.RELEASE</version>
    </parent>
    <groupId>com.bobo</groupId>
    <artifactId>stream-sender</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</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>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

1.3 配置文件

??配置文件中除了必要的服务名称，端口和Eureka的信息外我们还要添加RabbitMQ的注册信息

spring.application.name=stream-sender
server.port=9060
#设置服务注册中心地址，指向另一个注册中心
eureka.client.serviceUrl.defaultZone=http://dpb:[email protected]:8761/eureka/,http://dpb:[email protected]:8761/eureka/

#rebbitmq 链接信息
spring.rabbitmq.host=192.168.88.150
spring.rabbitmq.port=5672
spring.rabbitmq.username=dpb
spring.rabbitmq.password=123
spring.rabbitmq.virtualHost=/

1.4 创建消费发送者接口

??创建一个发送消息的接口。具体如下：方法名称自定义，返回类型必须是SubscribableChannel,在Output注解中指定交换器名称。

/**
 * 发送消息的接口
 * @author dengp
 *
 */
public interface ISendeService {

    /**
     * 指定输出的交换器名称
     * @return
     */
    @Output("dpb-exchange")
    SubscribableChannel send();
}

1.5 启动类

??在启动类中通过@EnableBinding注解绑定我们创建的接口类。

@SpringBootApplication
@EnableEurekaClient
// 绑定我们刚刚创建的发送消息的接口类型
@EnableBinding(value={ISendeService.class})
public class StreamSenderStart {

    public static void main(String[] args) {
        SpringApplication.run(StreamSenderStart.class, args);
    }
}

2.创建消息消费者服务

2.1 创建项目

2.2 pom文件

??添加的依赖和发送消息的服务是一致的

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.13.RELEASE</version>
    </parent>
    <groupId>com.bobo</groupId>
    <artifactId>stream-receiver</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</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>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

2.3 配置文件

??注意修改服务名称和端口

spring.application.name=stream-receiver
server.port=9061
#设置服务注册中心地址，指向另一个注册中心
eureka.client.serviceUrl.defaultZone=http://dpb:[email protected]:8761/eureka/,http://dpb:[email protected]:8761/eureka/

#rebbitmq 链接信息
spring.rabbitmq.host=192.168.88.150
spring.rabbitmq.port=5672
spring.rabbitmq.username=dpb
spring.rabbitmq.password=123
spring.rabbitmq.virtualHost=/

2.4 创建接收消息的接口

??此接口和发送消息的接口相似，注意使用的是@Input注解。

/**
 * 接收消息的接口
 * @author dengp
 *
 */
public interface IReceiverService {

    /**
     * 指定接收的交换器名称
     * @return
     */
    @Input("dpb-exchange")
    SubscribableChannel receiver();
}

2.5 创建处理消息的处理类

??注意此类并不是实现上面创建的接口。而是通过@EnableBinding来绑定我们创建的接口，同时通过@StreamListener注解来监听dpb-exchange对应的消息服务

/**
 * 具体接收消息的处理类
 * @author dengp
 *
 */
@Service
@EnableBinding(IReceiverService.class)
public class ReceiverService {

    @StreamListener("dpb-exchange")
    public void onReceiver(byte[] msg){
        System.out.println("消费者:"+new String(msg));
    }
}

2.6 启动类

??同样要添加@EnableBinding注解

@SpringBootApplication
@EnableEurekaClient
@EnableBinding(value={IReceiverService.class})
public class StreamReceiverStart {

    public static void main(String[] args) {
        SpringApplication.run(StreamReceiverStart.class, args);
    }
}

3.编写测试代码

??通过单元测试来测试服务。

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;

import com.bobo.stream.StreamSenderStart;
import com.bobo.stream.sender.ISendeService;

@RunWith(SpringRunner.class)
@SpringBootTest(classes=StreamSenderStart.class)
public class StreamTest {

    @Autowired
    private ISendeService sendService;

    @Test
    public void testStream(){
        String msg = "hello stream ...";
        // 将需要发送的消息封装为Message对象
        Message message = MessageBuilder
                                .withPayload(msg.getBytes())
                                .build();
        sendService.send().send(message );
    }
}

启动消息消费者后，执行测试代码。结果如下:

消息接收者获取到了发送者发送的消息，同时我们在RabbitMQ的web界面也可以看到相关的信息

总结

??我们同stream实现了消息中间件的使用，我们发现只有在两处地址和RabbitMQ有耦合，第一处是pom文件中的依赖，第二处是application.properties中的RabbitMQ的配置信息，而在具体的业务处理中并没有出现任何RabbitMQ相关的代码，这时如果我们要替换为Kafka的话我们只需要将这两处换掉即可，即实现了中间件和服务的高度解耦。

原文地址：https://www.cnblogs.com/dengpengbo/p/11103943.html

时间： 2024-10-02 01:25:08

Spring cloud stream【入门介绍】的相关文章

Spring Cloud Stream教程（五）编程模型

本节介绍Spring Cloud Stream的编程模型.Spring Cloud Stream提供了许多预定义的注释,用于声明绑定的输入和输出通道,以及如何收听频道. 声明和绑定频道 触发绑定@EnableBinding 您可以将Spring应用程序转换为Spring Cloud Stream应用程序,将@EnableBinding注释应用于应用程序的配置类之一.@EnableBinding注释本身使用@Configuration进行元注释,并触发Spring Cloud Stream基础架构

Spring Cloud Stream消费失败后的处理策略（三）：使用DLQ队列（RabbitMQ）

应用场景 前两天我们已经介绍了两种Spring Cloud Stream对消息失败的处理策略: 自动重试:对于一些因环境原因(如:网络抖动等不稳定因素)引发的问题可以起到比较好的作用,提高消息处理的成功率. 自定义错误处理逻辑:如果业务上,消息处理失败之后有明确的降级逻辑可以弥补的,可以采用这种方式,但是2.0.x版本有Bug,2.1.x版本修复. 那么如果代码本身存在逻辑错误,无论重试多少次都不可能成功,也没有具体的降级业务逻辑,之前在深入思考中讨论过,可以通过日志,或者降级逻辑记录的方式把错

Spring Cloud Stream消费失败后的处理策略（四）：重新入队（RabbitMQ）

应用场景 之前我们已经通过<Spring Cloud Stream消费失败后的处理策略(一):自动重试>一文介绍了Spring Cloud Stream默认的消息重试功能.本文将介绍RabbitMQ的binder提供的另外一种重试功能:重新入队. 动手试试 准备一个会消费失败的例子,可以直接沿用前文的工程,也可以新建一个,然后创建如下代码的逻辑: @EnableBinding(TestApplication.TestTopic.class) @SpringBootApplication pub

（十七）JAVA springcloud ssm b2b2c多用户商城系统-消息驱动 Spring Cloud Stream

在使用spring cloud云架构的时候,我们不得不使用Spring cloud Stream,因为消息中间件的使用在项目中无处不在,我们公司后面做了娱乐方面的APP,在使用spring cloud做架构的时候,其中消息的异步通知,业务的异步处理都需要使用消息中间件机制.spring cloud的官方给出的集成建议(使用rabbit mq和kafka),我看了一下源码和配置,只要把rabbit mq集成,kafka只是换了一个pom配置jar包而已,闲话少说,我们就直接进入配置实施: 1. 简

第十章 消息驱动的微服务： Spring Cloud Stream

Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架. 它可以基于Spring Boot 来创建独立的. 可用于生产的 Spring 应用程序. 它通过使用 Spring Integration 来连接消息代理中间件以实现消息事件驱动. Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现,并且引入了发布-订阅. 消费组以及分区这三个核心概念. 简单地说, Spring Cloud Stream 本质上就是整合了 Spr

Spring cloud stream【消息分组】

??上篇文章我们简单的介绍了stream的使用,发现使用还是蛮方便的,但是在上个案例中,如果有多个消息接收者,那么消息生产者发送的消息会被多个消费者都接收到,这种情况在某些实际场景下是有很大问题的,比如在如下场景中,订单系统我们做集群部署,都会从RabbitMQ中获取订单信息,那如果一个订单同时被两个服务获取到,那么就会造成数据错误,我们得避免这种情况.这时我们就可以使用Stream中的消息分组来解决了! Stream消息分组 ??消息分组的作用我们已经介绍了.注意在Stream中处于同一个gr

Spring cloud stream【消息分区】

??在上篇文章中我们给大家介绍了Stream的消息分组,可以实现消息的重复消费的问题,但在某些场景下分组还不能满足我们的需求,比如,同时有多条同一个用户的数据,发送过来,我们需要根据用户统计,但是消息被分散到了不同的集群节点上了,这时我们就可以考虑消息分区了. ??当生产者将消息数据发送给多个消费者实例时,保证同一消息数据始终是由同一个消费者实例接收和处理. Stream 消息分区 创建项目 ??将我们上篇文章中的分组的三个项目,拷贝一份修改名称及服务名称 没有分区的情况下演示 发送多条消息查看

Spring Cloud Stream异常处理

应用处理 当消费者在处理接收到的消息时,有可能会由于某些原因而抛出异常.若希望对抛出来的异常进行处理的话,就需要采取一些异常处理手段,异常处理的方式可分为三种:应用层面的处理.系统层面的处理以及通过RetryTemplate进行处理. 本小节先来介绍较为常用的应用层面的异常处理方式,该方式又细分为局部处理和全局处理. 局部处理 Stream相关的配置内容如下: spring: cloud: stream: rocketmq: binder: name-server: 192.168.190.12

Spring Cloud Stream如何消费自己生产的消息

在上一篇<Spring Cloud Stream如何处理消息重复消费>中,我们通过消费组的配置解决了多实例部署情况下消息重复消费这一入门时的常见问题.本文将继续说说在另外一个被经常问到的问题:如果微服务生产的消息自己也想要消费一份,应该如何实现呢? 常见错误 在放出标准答案前,先放出一个常见的错误姿势和告警信息(以便您可以通过搜索引擎找到这里^_^).以下错误基于Spring Boot 2.0.5.Spring Cloud Finchley SR1. 首先,根据入门示例,为了生产和消费消息,需