SpringCloud Stream

我不是女神ヾ 2021-10-19 01:12 377阅读 0赞

## 一、SpringCloud Stream简介 ##

> Spring Cloud Stream是spring数据集成的一个组成部件，为开发人员提供了更加简易的与外部系统连接的方式。  
> Spring Cloud Stream对消息中间件提供了进一步的封装，可以做到代码层面无感知的与中间件交互。甚至可以做到动态切换中间件组件。（如：RabbitMQ和Kafka的切换）  
> 使用Spring Cloud Stream开发，可以让微服务开发进一步解耦，让服务开发人员将注意力集中在业务逻辑的处理上。

## 二、Spring Cloud Stream结构简图 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbmluZ2tpZA_size_16_color_FFFFFF_t_70]

> Inputs - 代表从外部中间件读取数据，输入到应用中。  
> Outputs - 代表从应用中写出数据，输出到外部中间件。  
> Binder - 是应用于中间件的封装，可以通过Binder快速与中间件实现数据交互，可以动态切换中间件，可以通过配置定制中间件相关信息。  
> Middleware - 中间件组件，通常代表RabbitMQ或Kafka。

## 3. SpringCloud Stream应用 ##

> 使用Stream做微服务开发，需要依赖stream相关启动器，如：通过stream访问RabbitMQ需要依赖spring-cloud-starter-stream-rabbit

1.  添加依赖

<dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>

1.  添加相关配置

spring.rabbitmq.host=192.168.1.122
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=test
    spring.rabbitmq.password=123456
    # 连接访问的RabbitMQ虚拟主机路径。默认为'/'。
    spring.rabbitmq.virtualHost=/

1.  创建消息生产者 - Producer

> 消息提供者不需要关注与RabbitMQ耦合的相关操作，所有操作都针对于spring-cloud-stream组件。让代码和具体的消息中间件弱耦合。

/**
     * 使用spring-cloud-stream开发消息的提供者，不需要提供接口定义的实现。
     * Spring Cloud会提供一个接口的动态代理对象。
     * 后续代码中对消息的发送处理是通过SubscribableChannel信道对象实现的。
     */
    public interface IFirstMessageSender {
    
    	/**
    	 * @Output - 绑定RabbitMQ中的Exchange，属性value为Exchange的命名。
    	 *  Exchange的种类由spring-cloud-stream管理，默认使用topic。
    	 *  默认使用的Exchange是订阅发布类型的，发送到RabbitMQ中的消息会让所有对应的消息消费者处理。
    	 * @return SubscribableChannel 信道对象。用于实现数据输出的具体对象。
    	 */
    	@Output("test-stream")
    	SubscribableChannel sendMessage();
    }

> 消息提供者应用启动类需要提供新注解@EnableBinding，用于绑定处理消息相关逻辑接口。起到通知spring容器为接口准备动态代理对象的目的。

/**
     * @EnableBinding 用于绑定消息提供者或消息消费者的注解。
     *  用于通知spring cloud为对应的接口提供动态代理对象
     */
    @SpringBootApplication
    @EnableEurekaClient
    @EnableBinding(value={IFirstMessageSender.class})
    public class StreamProducerApplication {
    	public static void main(String[] args) {
    		SpringApplication.run(StreamProducerApplication.class, args);
    	}	
    }

> 在其他代码中，如果需要处理消息，则直接注入接口IFirstMessageSender即可。处理过程全部代码与具体消息中间件无关，都通过spring-cloud-stream相关API实现

@Controller
    public class MessageController {
    	@Autowired
    	private IFirstMessageSender sender ;
    	
    	/**
    	 * 通过IFirstMessageSender发送消息到RabbitMQ。
    	 * @param message 要发送的消息内容。
    	 * @return
    	 */
    	@RequestMapping(value="/sendMsg", produces={"application/json;charset=UTF-8"})
    	@ResponseBody
    	public String sendMsg(String message){
    		// withPayload方法参数类型为Object，即可以发送任意类型数据的消息。
    		// 要求要传递的消息数据对象必须可序列化。
    		Message<String> m = MessageBuilder.withPayload(message).build();		
    		this.sender.sendMessage().send(m);
    		return "{\"status\":\"OK\"}";
    	}	
    }

1.  创建消息的消费者 - Consumer

> 消息消费者的开发同样针对于spring-cloud-stream相关API，不需要与具体消息中间件耦合。

public interface IFirstMessageReciver {
    	/**
    	 * @Input - 绑定RabbitMQ中的Exchange，属性value为Exchange的命名。
    	 *  如果和@Output注解的value属性相同，代表当前消息消费者处理对应消息提供者发送的消息。
    	 * @return SubscribableChannel 信道
    	 */
    	@Input("test-stream")
    	SubscribableChannel recive();	
    }

> 在处理消息的时候，需要具体的处理逻辑，消息数据通过spring-cloud-stream获得，不需要通过自定义开发连接消息中间件获取消息。

@Service
    @EnableBinding(value={IFirstMessageReciver.class})
    public class MessageService {
    	/**
    	 * @StreamListener - 绑定监听到指定的Exchange。
    	 * @param message 接收到的消息内容。此参数类型必须是可序列化的。
    	 */
    	@StreamListener("test-stream")
    	public void onMessage(String message){
    		System.out.println("recive message content : " + message);
    	}
    }

> 启动类同样需要@EnableBinding描述

/**
     * @EnableBinding 用于绑定消息提供者或消息消费者的注解。
     *  用于通知spring cloud为对应的接口提供动态代理对象
     */
    @SpringBootApplication
    @EnableEurekaClient
    @EnableBinding(value={IFirstMessageReciver.class})
    public class StreamConsumerApplication {
    	public static void main(String[] args) {
    		SpringApplication.run(StreamConsumerApplication.class, args);
    	}	
    }

1.  总结

> 在默认环境下，spring-cloud-stream使用的RabbitMQ中的Exchange是topic类型的，Producer发送的消息会被所有的对应Consumer同时处理。且默认创建的队列都是Auto-Delete的。这里的Consumer集群每个节点会监听一个Queue，Queue的名称是Exchange名称.随机后缀名，所以每个Consumer都在处理不同的Queue。而Topic类型的Exchange会根据路由键来分发消息，其路由键的匹配规则为Exchange名称.\*。所以同一个消息会发送到所有匹配规则的Queue中。  
> 通过spring-cloud-stream访问消息中间件可以达到开发无感知，保证代码和具体的中间件容器解耦。开发者可以更加关注业务逻辑，而不是要处理的中间件逻辑。

## 四、消息分组 ##

> 使用消息分组可以达到消息点对点传递的目的，且可以将队列变更为持久队列，避免消息丢失。

1.  创建消息的提供者 - Producer， 并创建配置文件, 提供分组配置

1.  在代码中，@Output注解中的value属性必须和配置文件对应。

@Output("outputName")
    SubscribableChannel sendMessage();

1.  创建消息消费者 - Consumer, 并创建配置文件

> 在代码中，@Input注解中的value属性和@StreamListener注解中的value属性必须和配置文件对应。

@Input("inputName")
    	SubscribableChannel recive();
    	@StreamListener("inputName")
    	public void onMessage(String message){
    		System.out.println("recive message content : " + message);
    	}

1.  总结

> 所谓分组，使用的Exchange类型仍旧是topic，只是将Consumer集群注册到同一个Queue上，这样在Queue发生变化时，Consumer集群会轮训处理Queue中的消息，保证消息不被重复处理。  
> 分组后，使用的队列是持久化队列。不会因为Consumer全部关闭而自动删除。可以有效避免消息丢失。

## 五、消息分区 ##

> 使用消息分区可以实现相同的消息一定发送给同一个Consumer节点处理。是开发中为避免队列中有重复消息被不同Consumer处理的情况。

1.  创建消息的提供者 - producer，并创建配置文件，提供分区配置

# 定义分组信息。格式为：
    # spring.cloud.stream.bindings.自定义名称.destination=Exchange名称
    # 其中自定义名称是在代码中的@Output注解中使用的。
    # Exchange名称是用于定义当前Producer发送消息对应的Exchange。
    spring.cloud.stream.bindings.outputName.destination=test-exchange
    # 定义分区信息。分区配置必须配合分组配置，分区操作是分组操作的进阶。
    # 配置格式：
    # spring.cloud.stream.bindings.自定义名称.producer.partitionKeyExpression=payload
    # 代表对应的Exchange分区的匹配表达式。
    # payload代表根据内容分区。
    spring.cloud.stream.bindings.outputName.producer.partitionKeyExpression=payload
    # partitionCount用于配置分区数量，根据具体的业务需求配置。
    spring.cloud.stream.bindings.outputName.producer.partitionCount=2

1.  在代码中，@Output注解中的value属性必须和配置文件对应。

@Output("outputName")
    	SubscribableChannel sendMessage();

1.  创建消费者 - consumer， 并创建配置文件

# spring.cloud.stream.bindings.自定义名称.destination=Exchange名称
    spring.cloud.stream.bindings.inputName.destination=test-exchange
    # spring.cloud.stream.bindings.自定义名称.group=队列后缀名
    # 定义分组实质上就是指定队列命名，具体队列名称为Exchange名称.队列后缀名
    spring.cloud.stream.bindings.inputName.group=queue-group
    # 开启消费者分区控制
    spring.cloud.stream.bindings.inputName.consumer.partitioned=true
    # 定义分区数量
    spring.cloud.stream.instanceCount=2
    # 定义当前Consumer分区编号，编号从0开始，自然数升序排列
    spring.cloud.stream.instanceIndex=0

1.  在代码中，@Input注解中的value属性和@StreamListener注解中的value属性必须和配置文件对应。

1.  总结

> 分区操作并不是RabbitMQ提供的，是由Spring Cloud来控制的。分区操作使用的Exchange类型仍旧是Topic。  
> 分区控制是根据路由键routing-key实现的。Spring-cloud-stream在发送消息之前，会判断payload是否曾经发送过，如果发送过，则使用曾经使用的那个routing-key。如果未发送过，则可以随机生成有效的routing-key发送。  
> 对效率有一定的影响，影响很小。是毫秒级别。在Spring-cloud-stream中用于记录payload是否重复的方式是ConcurrentHashMap。检索payload是否重复的效率还是非常高的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbmluZ2tpZA_size_16_color_FFFFFF_t_70]: /images/20211018/20d9c12816ba4160ba0185f802278235.png