SpringCloudStream实战拆解以及3.1后新版本特性分析

Posted roykingw

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了SpringCloudStream实战拆解以及3.1后新版本特性分析相关的知识,希望对你有一定的参考价值。

文章目录

一体化MQ解决方案-SpringCloudStream 深度解析
楼兰:你的神秘java宝藏
老友聊IT,老王多指教

一、MQ的水有多深

​ MQ:也就是MessageQueue,消息对列。对列,是一种具有FIFO先进先出特性的数据结构。在互联网中,基于MQ的数据结构,人们封装出了非常多各具特色的分布式消息中间件产品,由此构建出了非常多很有特色的应用场景。例如像我们经常用的QQ和微信,就是很典型的消息中间件。消息可以由生产者发送到MQ中进行排队,然后消息的消费者可以对消息按照顺序进行处理。而在网络应用层面,消息中间件更是实现分布式消息驱动场景的利器。

​ 在当今互联网中,MQ已经不是一个神秘的东西了,关于MQ的作用主要有三个:

  • 异步

    例子:快递员送快递,早期直接送到客户家签收,这样效率会很低。引入菜鸟驿站后,快递员只需要将快递放到菜鸟驿站,就可以继续派送其他快递去了。客户可以按照自己的时间安排去菜鸟驿站取快递。

    作用:异步能提高系统的响应速度、吞吐量。

  • 解耦

    例子:《Thinking in JAVA》很经典,但是都是英文,我们看不懂。这时引入编辑社,可以将文章翻译成其他语言,这样就可以完成英语与其他语言的交流。

    作用:1、服务之间接口,可以减少服务之间的影响,提升应用的稳定性以及可扩展性。

    2、接口后可以实现数据分发。生产者发送一个消息后,可以由一个或者多个消费者进行消费,并且消费者的增加或者减少对生产者没有影响。

  • 削峰

    例子:长江每年都会涨水,但是下游出水口的速度是基本稳定的,所以经常会出现水灾。引入三峡大坝后,可以把高峰期的水储存起来,下游慢慢排水。

    所用:让应用可以以稳定的系统资源应对突然的流量冲击。

​ 目前主流的MQ产品有Kafka、RabbitMQ、RocketMQ三个。我们对三个产品做下简单的比较,重点要理解他们的适用场景。

​ 这三种主流产品都非常成熟,也都有各自不同的一些特性功能,适用于不同的业务场景。但是,这会不会给你带来困惑?同样都是MQ,不同场景下要使用不同的产品,而每个产品都有自己的客户端。那岂不是在细分场景下要用好MQ,就需要一套庞大的技术体系?

​ 有没有一套框架可以像用JDBC去连接各种数据库一样,用一套快速简单的标准对接各种各样的MQ产品呢?这就是我们这次要聊到的SpringCloudStream框架(后续简称SCS)。

二、一套框架,盲拆所有MQ产品

​ SpringCloudStream是SpringCloud家族的一个重要组件。官网地址: https://spring.io/projects/spring-cloud-stream 他是一个基于共享MQ系统,用于构建高度可伸缩的,事件驱动的问服务应用的框架。简单粗暴的理解就是,一个统一的框架标准连接所有的MQ产品。在官网可以看到SCS已经集成了非常多的MQ产品。

​ 我估计你跟我一样,这么多MQ产品,有些连听都没听说过,更别说在系统中用上了。但是有了SCS后,这些就都不用担心了。SCS是如何帮我们连接所有MQ产品的呢?接下来我们就用常见的Kafka,RabbitMQ和RocketMQ为例,开始一段舒心的SCS之旅。

注:以下部分,假定你对于Kafka、RabbitMQ和RocketMQ已经有了初步了解,并且搭建有本地服务。

各组件的版本如下:RabbitMQ版本:3.9.5, RocketMQ版本:4.9.1, Kafka版本:3.0.1。

2.1:一分钟快速搭建SCS连接RabbitMQ环境。

​ SCS是基于SpringBoot构建的,首先我们创建一个基于Maven的SpringBoot项目。按照Maven应用三板斧,快速搭建基础环境。

注:这里先快速把开发环境构建起来,不用理解每一步的具体含义。后面会做深入分析

1、pom.xml 依赖配置

​ 按照最常用的开发方式引入SpringBoot以及SpringCloud的集成版本控制,并且引入RabbitMQ的SCS集成依赖包。

	<!-- 引入SpringBoot与SpringCloud的集成版本 -->
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>2.3.4.RELEASE</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Hoxton.SR6</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <!-- 引入RabbitMQ的SCS集成依赖 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <!-- artifactId>spring-cloud-starter-stream-rabbit</artifactId -->
            <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

2、启动类 引入核心注解

​ 注意在启动类中引入@EnableBinding注解。

@SpringBootApplication
@EnableBinding(Source.class, Sink.class)
public class SCSApplication 

    public static void main(String[] args) 
        SpringApplication.run(SCSApplication.class);
    

3、application.properties 配置文件

​ 在配置文件中,只需要简单的目标配置,甚至都不需要指定MQ的服务地址。

spring.cloud.stream.bindings.output.destination=scstreamExchange

spring.cloud.stream.bindings.input.destination=scstreamExchange
spring.cloud.stream.bindings.input.group=stream
spring.cloud.stream.bindings.input.content-type=text/plain

​ 这样,基础的开发框架就搭建成功了。接下来搭建消息发送者和消息生产者。

4、创建一个发送消息的Controller

@RestController
public class SendMessageController 
    @Autowired
    private Source source;

    @GetMapping("/send")
    public Object send(String message) 
		MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(message);
        source.output().send(messageBuilder.build());
        return "message sended : "+message;
    

5、声明一个消息消费者

@Component
public class MessageReceiver 

    @StreamListener(Sink.INPUT)
    public void process(Object message) 
        System.out.println("received message : " + message);
    

​ 到这里,整个MQ应用环境就搭建完成了。接下来可以启动应用,然后访问测试接口发送消息。http://localhost:8080/send?message=123 。

当然,要保证本地的RabbitMQ服务正常启动了。

如果你对RabbitMQ有点基础的了解,也可以去RabbitMQ的管理控制台确认一下,这个简单的应用程序是如何实现RabbitMQ的消息收发的。

​ 实际上他的实现过程就是在本地的RabbitMQ中创建了一个scstreamExchange的交换机。应用中的Source组件就是往这个交换机中发送消息。然后,在scstreamExchange这个交换机上,也声明并绑定了一个队列,scstreamExchange.stream队列。应用中的消息监听器Sink组件,就是消费这个队列上的消息。

​ 当然,如果你对RabbitMQ一点也不了解,那也没关系,至少通过这个不超过十行代码的应用,就可以完成与RabbitMQ的消息交互。你只需要知道,这个应用当中的Source消息生产者和Sink消息消费者,是解耦的。在实际应用中,可以将他们拆分到不同的应用当中,实现消息的异步消费。这就够了。

你不需要理解与RabbitMQ交互的细节,甚至连RabbitMQ服务地址,SCS中都提供了默认的配置。你所要关注的,只有业务。

2.2: 一秒切换对接kafka

​ 如果只是简化与RabbitMQ的交互过程,那么还体现不出SCS的强大。如果需要将上面的简单应用改为对接Kafka,那么整个业务代码不需要做任何改动,只需要修改pom.xml中的maven依赖,将RabbitMQ的SCS依赖替换成Kafka的SCS依赖即可。

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

​ 接下来,重新启动示例应用,再次访问发送消息接口。消息就会改由Kafka来转发。

​ 同样,如果你熟悉Kafka产品,也能看到实际上SCS框架在Kafka上申请了一个scstreamExchange的Topic,并对这个Topic声明了一个消费者。

2.3:一秒切换对接RocketMQ

​ 接下来对接RocketMQ就简单了,同样是修改pom.xml中的maven依赖,代码不需要做任何改动。

	<dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            <version>2021.1</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-acl</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-client</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-acl</artifactId>
            <version>4.9.1</version>
        </dependency>

从这个maven依赖中就能看出, kafka和RabbitMQ是亲儿子,RocketMQ是厂商自己拉扯大的。所以客户端版本匹配需要注意下。

​ 使用Maven重新编译后,就可以重新访问发送消息的接口。这次消息就会通过RocketMQ发送给后端。

​ 同样如果你了解RocketMQ产品,那么可以去RocketMQ的Console控制台上查看他的实现机制。自动创建了一个scstreamExchange的Topic,并声明了一个stream消费者组。

​ 在这个简单的实验过程中可以看到,SCS框架以一种标准化的方式提供了与MQ对接的功能,而与具体的MQ产品交互的细节,都被隐藏在框架之中。在这个过程中,如果你对相关MQ产品比较熟悉,那么可以大致揣测出他的实现方式。但是,就算你对这些MQ产品一无所知,通过SCS框架,也完全不影响基础的编程开发。这正是SCS框架的强大之处。

三、详解SCS三神器

​ SCS框架抽象出了一个非常简洁的MQ模型,来对接各种MQ产品。这个模型其实非常简单,只包含以下三个组件:

  • Destination Binder: 负责集成外部消息系统的组件。
  • Destination Binding: 由Binder创建,负责沟通外部消息系统、消息发送者和消息消费者的桥梁。
  • Message:消息发送者与消息消费者沟通的简单数据结构。

​ 可以看到,这个模型非常简单,使用时也非常方便。但是简单,也意味着SCS中的这些概念模型,与实际MQ产品中的概念是有比较大的区别的。例如,在RabbitMQ中有Exchange交换机、Queue队列等这些概念,但是在SCS中并没有对应的概念。在使用具体产品时,需要非常注意这些模型之间的对应关系。

​ 接下来会以RabbitMQ作为示例,分别了解这些核心概念。

3.1 Binder

​ SCS通过Binder定义一个外部消息服务器。默认情况下,SCS会使用对应的SpringBoot插件来构建Binder。所以,对于RabbitMQ来说,SCS默认就支持spring-boot-starter-amqp组件中提供的RabbitMQ的服务配置信息。这些配置信息,在application.properties配置文件中,都以spring.rabbitmq开头。而这一组配置,都是有默认值的。

spring.rabbitmq.host=local
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

​ 如果你熟悉RabbitMQ就会知道,这其实就是一组默认的RabbitMQ服务。这也就是为什么在之前的示例中,都没有指定服务器地址,但是SCS也能正常工作的原因。

对于Kafka和RocketMQ,也同样会使用默认的服务。

​ 另外,在SCS中,也可以支持按照自己的方式配置多个Binder访问不同的外部消息服务器。这些配置的方式都是通过spring.cloud.stream.binders.[bindername].environment.[props]=[value]的方式进行配置。另外,如果配置了多个Binder,也可以通过spring.cloud.stream.default-binder属性指定默认的Binder。例如:

spring.cloud.stream.binders.testbinder.environment.spring.rabbitmq.host=localhost
spring.cloud.stream.binders.testbinder.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.testbinder.environment.spring.rabbitmq.username=guest
spring.cloud.stream.binders.testbinder.environment.spring.rabbitmq.password=guest
spring.cloud.stream.binders.testbinder.environment.spring.rabbitmq.virtual-host=/
spring.cloud.stream.default-binder=testbinder

​ 通过这个方式,就配置了一个名为testbinder的Binder,并且将他指定成了默认的Binder。environment后面的这些props就是跟具体MQ产品相关的属性了。

3.2 Binding

​ Binding是SCS中实际进行消息交互的桥梁。在SCS中,就是通过Binding和Binder建立绑定关系,然后客户端就只需要通过Binding进行实现的消息收发了。

​ 在SCS框架中,配置Binding需要首先对他进行声明。当前版本的SCS框架中,声明Binding的方式是在应用中引入@EnableBinding注解。引入这个注解后,应用会向Spring容器中注入一个Binding接口的实例对象。在SCS中,默认提供了Source、Sink、Process三个接口对象,分别代表消息的生产者、消费者和中间处理者。这三个对象都是简单的接口,可以直接拿来使用。

//消息生产者
public interface Source 
	String OUTPUT = "output";//Binding对象名
	@Output(Source.OUTPUT)
	MessageChannel output(); //发送消息工具

//消息消费者
public interface Sink 
	String INPUT = "input"; //Binding对象名
	@Input(Sink.INPUT)
	SubscribableChannel input();//消息接收工具

//即是发送者,又是接收者。代表一个中间处理步骤
public interface Processor extends Source, Sink 


​ 当然,如果这些默认的接口不够用,你也可以照样声明自己的接口,通过自己的接口声明新的Binding对象。

​ 接下来,就需要在Spring应用的某一个合适的位置通过@EnableBinding注解声明这个Binding对象。例如最为常见的是放到启动类上声明。

@SpringBootApplication
@EnableBinding(Source.class, Sink.class)//声明Binding
public class SCSApplication 
    public static void main(String[] args) 
        SpringApplication.run(SCSApplication.class);
    


​ 在示例中,通过Source和Sink接口,就声明出了两个Binding,对象名分别是output和input。

​ 完成声明后的Binding对象,就可以直接来用了。比如通过source.output()方法获取MessageChannel对象,进而发送消息。然后通过@StreamListener(Sink.INPUT)注解声明消息消费者。使用方式在示例中有,就不再过多赘述。

​ 但是,默认的Binding对象由于缺少很多关键的属性配置,在某些环境下会出现很多问题。比如,在RabbitMQ中,如果不经任何配置,就会声明出一个默认的Exchange和Queue。但是默认的名字会非常奇怪,而且很多细节功能都不好用。所以,通常都要对这个Binding进行配置。配置的方式都是在application.properties中配置。这些配置都是按照spring.cloud.stream.bindings.[bindingname].[props]=[value]这样的格式指定的。例如:

spring.cloud.stream.bindings.output.destination=scstreamExchange
#指定binder。
spring.cloud.stream.bindings.output.binder=testbinder

spring.cloud.stream.bindings.input.destination=scstreamExchange
spring.cloud.stream.bindings.input.group=stream
spring.cloud.stream.bindings.input.content-type=text/plain
#指定binder。
spring.cloud.stream.bindings.input.binder=testbinder

​ 其中,通过binder属性可以指定对应的Binder。如果不指定的话,就会使用默认的Binder。通过这样的简单配置,就指定了output这个Binding对应的Exchange交换机是scstreamExchange,input也对应这同一个交换机,所属的消费者组是stream。

​ 关于消费者组,如果你熟悉Kafka或者RocketMQ,相信你就不会陌生。同属一个组的多个消费者实例,会共同消费同一份消息副本。但是在RabbitMQ中,却是没有消费者组这个概念的。接下来,应用启动后,在RabbitMQ中,就会声明一个scstreamExchange,并且在这个交换机上绑定一个scstreamExchange.stream的队列。这样,当有多个消费者实例启动时,这些消费者实例都会一起绑定在这个队列上,这样同一个消费者组的多个示例,还是会共同消费这一个队列上的同一个消息副本,从而变相的实现了分组消费机制。

​ 这里的这些属性是SCS中通用的配置属性,对所有支持的MQ产品都会生效。例如:

spring.cloud.stream.bindings.input.consumer.max-attempts=3

​ 这个配置属性就可以指定input这个消费者的消息最大重试次数。默认值是3。通过SCS框架,就以一种统一的方式定义了所有支持的MQ产品中的消息重试次数,尽管在不同的MQ产品中实现消息重试的配置方式是不一样的。

3.3 Message

​ SCS中,有了Binder和Binding之后,就相当于有了完整的消息传输通道了。接下来就可以传递消息了。但是不同的MQ产品中对于消息的定义其实也是不相同的,例如RocketMQ中的消息有tag,key,自定义属性等。而在RabbitMQ的消息中也有messageId,headers头信息等。SCS框架就需要对这些消息类型进行统一。统一后的消息类型是这样的:

public interface Message<T> 
    T getPayload();
    MessageHeaders getHeaders();


​ 其中Payload就是消息实体,在SCS中定义成了一个泛型,也就是说可以直接传递对象。MessageHeaders就是消息的头部属性,可以认为是消息的补充属性。这样,不同的MQ产品下,就可以通过不同的MessageHeaders属性来代表各自的消息差异,而消息内容就可以通过Payload统一。例如如果你熟悉RabbitMQ产品的话,就知道,RabbitMQ中有一个非常重要的概念routingKey。通过routingKey可以定制Exchange与Queue之间的路由关系。这个routingKey在RabbitMQ的实现包中就可以通过在Headers当中指定一个routingkey属性来实现。

MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(message).setHeader("routingkey","info");

​ Message接口下,实际上也是有很多实现类的,简单看一下MessageBuilder的build方法,就能看到,这种方式是可以构建ErrorMessage和GenericMessage两种不同的消息的。

    public Message<T> build() 
        if (!this.modified && !this.headerAccessor.isModified() && this.originalMessage != null && !this.containsReadOnly(this.originalMessage.getHeaders())) 
            return this.originalMessage;
         else 
            return (Message)(this.payload instanceof Throwable ? new ErrorMessage((Throwable)this.payload, this.headerAccessor.toMap()) : new GenericMessage(this.payload, this.headerAccessor.toMap()));
        
    

​ 有了Binder,Binding和Message这三个工具后,你就可以直接用SCS框架去对接所有支持的MQ产品了。

四、SCS提供的高级功能

4.1 消息分组消费机制

​ SCS框架不光是统一与各个MQ产品之间的交互方式,还给各个MQ之间设计了一套统一的消费机制,就是分组消费机制。他的基础思想是在生产者实例和消费者实例之间建立一种对应关系,生产者实例发出的消息只会被对应的消费者消费,这样,当应用中有多个生产者实例和多个消费者实例时,就可以建立一种类似于Kafka和RocketMQ的分组消费的机制。

​ 例如,在RabbitMQ的示例应用中,对于生产者output,可以增加如下分区配置:

spring.cloud.stream.bindings.output.destination=scstreamExchange
#指定参与消息分区的消费端节点数量
spring.cloud.stream.bindings.output.producer.partition-count=2
#只有消费端分区ID为1的消费端能接收到消息
spring.cloud.stream.bindings.output.producer.partition-key-expression=1

​ 这就表示当前生产者会有两个分区实例,当前分区实例的分区键是1。

​ 然后,对于消费者input,如果增加以下分区属性:

spring.cloud.stream.bindings.input.destination=scstreamExchange
spring.cloud.stream.bindings.input.group=myinput
#启动消费分区 新版本这个属性已经取消,改为由分区表达式自动判断
spring.cloud.stream.bindings.input.consumer.partitioned=true
#参与分区的消费端节点个数
spring.cloud.stream.bindings.input.consumer.instance-count=2
#设置该实例的消费端分区ID
spring.cloud.stream.bindings.input.consumer.instance-index=1

​ 这也表示当前消费者组会有两个分区实例,当前分区实例的分区键是1。

​ 这时,把应用启动起来,消费者发送的消息,依然能够正常被消费者消费到。但是,如果将消费者的instance-index属性改成0,或者是2,那重启应用后,就无法消费到生产者发出的消息了。也就是说,只有与生产者的分区键匹配的消费者,才能接收到生产者发送过来的消息。如果分区键不匹配,那就接收不到。这里要注意,这并不是表示消息丢失了,而是当前的消费者实例和生产者实例没有对应上,就无法消费对应的消息了。

​ 我们这个示例还太过简单,生产者和消费者的分区键都是直接指定的,不具备实际使用的价值。但是,SCS有更灵活的机制,让生产者可以根据消息本体来产生生产者的分区键,这样,就可以具备真正的分区消费功能了。

​ 这主要涉及到SCS中的producer后的两个关键的配置属性:partition-key-extractor-name和partition-selector-name。

​ 其中,partition-key-extractor-name属性需要指向一个实现了PartitionKeyExtractorStrategy接口的实现类,用来计算从Message中如何产生一个动态的分区Key。这个机制与partition-key-expression配置是互斥的。而partition-selector-name属性需要指向一个实现了PartitionSelectorStrategy接口的实现类,用来计算如何从分区Key中获取一个动态的分区键。

​ 例如,按照以下配置,就可以实现按照消息的轮询接收。即生产者发送消息的分区键会奇数偶数循环,而消费者只接收分区键为奇数的消息。

//step1、增加分区提取器
public class MyPartitionKeyExtractor implements PartitionKeyExtractorStrategy 
    public static final String PARTITION_PROP="partition";
    @Override
    public Object extractKey(Message<?> message) 
        return message.getHeaders().get(MyPartitionKeyExtractor.PARTITION_PROP);
    

//step2、增加分区键计算器
public class MyPartitionSelectorStrategy implements PartitionSelectorStrategy 
    @Override
    public int selectPartition(Object key, int partitionCount) 
        return Integer.parseInt(key.toString()) % partitionCount;
    

//step3、在application.properties中添加生产者的分区配置
spring.cloud.stream.bindings.output.destination=scstreamExchange
spring.cloud.stream.bindings.output.binder=testbinder
#指定参与消息分区的消费端节点数量
spring.cloud.stream.bindings.output.producer.partition-count=2
#只有消费端分区ID为1的消费端能接收到消息
#spring.cloud.stream.bindings.output.producer.partition-key-expression=1
#动态生成分区键
spring.cloud.stream.bindings.output.producer.partition-key-extractor-name=com.roy.partition.MyPartitionKeyExtractor
spring.cloud.stream.bindings.output.producer.partition-selector-name=com.roy.partition.MyPartitionSelector
//step4、发送消息时,给消息添加个动态增长的index属性。
    //index生成工具
    @Bean
    public AtomicInteger index()
        return new AtomicInteger(0);
    
//step5、在发送消息时,连续发送四条消息。
@RestController
public class SendMessageController 
    @Autowired
    private Source source;

    @Autowired
    private AtomicInteger index;

    @GetMapping("/sendPartitionBatch")
    public Object sendPartitionBatch(String message) 
        for (int i = 0; i < 4; i++) 
            MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(message).setHeader(MyPartitionKeyExtractor.PARTITION_PROP,index.incrementAndGet(干货 |《从Lucene到Elasticsearch全文检索实战》拆解实践

干货 |《从Lucene到Elasticsearch全文检索实战》拆解实践

基于SpringCloud的Microservices架构实战案例-架构拆解

Maven实战技巧「配置文件专题」全面的Setting文件的配置拆解说明介绍指南

Day698.Tomcat的日志框架及实战 -深入拆解 Tomcat & Jetty

一体化MQ解决方案-SpringCloudStream