码迷，mamicode.com
首页 > 编程语言 > 详细

Spring cloud stream【入门介绍】

时间：2019-06-28 18:16:37      阅读：132      评论：0      收藏：0      [点我收藏+]

标签：驱动   img   release   send   发布   receiver   channel   介绍   top   

案例代码: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:123456@eureka1:8761/eureka/,http://dpb:123456@eureka2: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:123456@eureka1:8761/eureka/,http://dpb:123456@eureka2: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的话我们只需要将这两处换掉即可，即实现了中间件和服务的高度解耦。

Spring cloud stream【入门介绍】

标签：驱动   img   release   send   发布   receiver   channel   介绍   top   

原文地址：https://www.cnblogs.com/dengpengbo/p/11103943.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！