Spring Cloud Stream消息处理机制学习笔记(二)
作者：互联网
消息发布场景：
用户信息一般不会发生变化，所以我们把信息放入缓存里，不再每一次都去查库，一旦用户信息发生变化，user_service会发布变更事件给到相关的订阅者并更新缓存信息。
一.实现消息发布：
主要实现Spring Cloud Stream创建Source组件，Binder组件的配置以及如何与user_service绑定；
pom依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>使用 @EnableBinding 注解
对于user_service而言，它是消息发布者，扮演Source角色；所以要给此spring boot应用绑上source组件，使用@EnableBinding；
@SpringCloudApplication
@EnableBinding(Source.class)
public class UserApplication {
public static void main(String[] args) {
SpringApplication.run(UserApplication.class, args);
}
}
@EnableBinding(Source.class) 该注解的作用是告诉spring cloud stream此应用是一个消息发布者，需要绑定到详细中间件。
定义事件
对于事件一般有通用的定义结构。事件类型，操作，事件模型
public class UserInfoChangedEvent{
//事件类型
private String type;
//操作
private String operation;
//模型
private User user;
}
创建Source
直接在使用时注入即可
@Component
public class UserInfoChangedSource {
private Source source;
@Autowired
public UserInfoChangedSource(Source source){
this.source = source;
}
private void publishUserInfoChangedEvent(String operation, User user){
UserInfoChangedEvent change = new UserInfoChangedEvent(
UserInfoChangedEvent.class.getTypeName(),
operation,
user);
source.output().send(MessageBuilder.withPayload(change).build());
}
}
构建了event事件，使用MessageBuilder方法把事件转换成MessageChannel中可传输的对象。调用out()方法放入管道MessageChannel，再调用管道的send()方法发送出去。
配置Binder
为了将消息发出去并且发送至目的地址，就需要配置相关属性。
spring:
cloud:
stream:
bindings:
output:
destination: userInfoChangedTopic
content-type: application/json
kafka:
binder:
zk-nodes: localhost
brokers: localhost
配置队列名称userInfoChangedTopic；kafka相关注册在Zookeeper 上的地址；
集成服务：
最后要做的事就是集成UserInfoChangedSource 到我们的user_service;只需要在使用的时候注入我们的UserInfoChangedSource类就可以了，因为之前已经加了@Component.
二.在服务中添加消息消费者
与消息发布一样需要依赖spring-cloud-stream、spring-cloud-starter-stream-kafka或者rabbitMq。并在启动类绑定Sink组件。
@SpringCloudApplication
@EnableBinding(Sink.class)
public class OrderApplication{
public static void main(String[] args) {
SpringApplication.run(InterventionApplication.class, args);
}
}
创建 Sink
负责具体处理消息的消费逻辑
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
public class UserInfoChangedSink {
@Autowired
private UserInfoRedis userInfoRedis;
@StreamListener("input")
public void handleChangedUserInfo(UserInfoChangedEvent userInfoChangedEvent) {
if(userInfoChangedEvent.getOperation().equals("ADD")) {
userInfoRedis.saveUser(userInfoChangedEvent.getUser());
} else if(userInfoChangedEvent.getOperation().equals("UPDATE")) {
userInfoRedis.updateUser(userInfoChangedEvent.getUser());
} else if(userInfoChangedEvent.getOperation().equals("DELETE")) {
userInfoRedis.deleteUser(userInfoChangedEvent.getUser().getUserName());
}
}
}
新的注解@StreamListener，将该注解添加到方法上就可以接收处理流中的事件。上述代码中@StreamListener("input"),意味着流经input通道的消息都将会交由此方法来处理。UserInfoRedis 根据自己业务情况去集成操作Redis的组件。
配置 Binder
配置Binder的方式和消息发布者几乎一样，如果发布采用的是默认通道output，那么接收消息的只需要将配置改为input
spring:
cloud:
stream:
bindings:
input:
destination: userInfoChangedTopic
content-type: application/json
kafka:
binder:
zk-nodes: localhost
brokers: localhost
三.Spring Cloud Stream 高级主题
创建一个自定义的通道接口类
public interface MyChannel {
String INPUT = "msg-input";
String OUTPUT = "msg-output";
@Output(OUTPUT)
MessageChannel output();
@Input(INPUT)
SubscribableChannel input();
}
默认的 Sink 通道接口
public interface Sink {
String INPUT = "input";
@Input("input")
SubscribableChannel input();
}
自定义类不仅有input还有output，因为消息的输出和输入需要配置。
创建消息自定义接收器
@EnableBinding(MyChannel.class)
public class MsgSink {
private static final Logger logger = LoggerFactory.getLogger(MsgSink.class);
@StreamListener(MyChannel.INPUT)
public void receive(Object payload){
logger.info("received:" + payload);
}
}
在 application.properties 中配置自定义通道的 input 和 output 的 destination ，使其绑定在一起
spring.cloud.stream.bindings.msg-input.destination=payload-topic
spring.cloud.stream.bindings.msg-output.destination=payload-topic
都绑定到payload-topic队列中，注意上面的配置bindings后面的配置这里就是MyChannel 接口自定义的通道名称；注意到 @Input 和 @Output 注解使用通道名称作为参数，如果没有名称，会使用带注解的方法名字作为参数。可以看出MyChannel 可以自定义任意多个input和output 。只有名称被注解使用到的才会生效。
消费者分组
在集群部署的系统中，希望服务的不同实例属于竞争关系，不应该重复消费同一消息，一个消息只能被服务集群的某一个实例处理。要想达到这个效果只需要将相同的服务放在同一消费者组内即可。唯一要做的事情也是重构Binder配置.配置如下
spring:
cloud:
stream:
bindings:
msg-input:
destination: payloadTopic
content-type: application/json
group: orderGroup
kafka:
binder:
zk-nodes: localhost
brokers: localhost
以上基于Kafka的配置信息中，“bindings”段中的通道名称使用了自定义的“msg-input”，并且在该配置项中设置了“group”为“orderGroup”。
消息分区
同时有多条同一个用户的数据，发送过来，我们需要根据用户统计，但是消息被分散到了不同的集群节点上了，这时我们就可以考虑消息分区了。
当生产者将消息数据发送给多个消费者实例时，保证同一消息数据始终是由同一个消费者实例接收和处理。
1.生产者配置
spring.application.name=stream-partition-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=/
# 对应 MQ 是 exchange outputProduct自定义的信息
spring.cloud.stream.bindings.outputProduct.destination=exchangeProduct
#通过该参数指定了分区键的表达式规则
spring.cloud.stream.bindings.outputProduct.producer.partitionKeyExpression=payload
#指定了消息分区的数量。
spring.cloud.stream.bindings.outputProduct.producer.partitionCount=2问题思考？分区建的表达式‘payload’ 是绑定的消息体计算出来的消费者实例分区index
2.消费者配置
实例A
spring.application.name=stream-partition-receiverA
server.port=9070
#设置服务注册中心地址，指向另一个注册中心
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=/
# 对应 MQ 是 exchange 和消息发送者的 交换器是同一个
spring.cloud.stream.bindings.inputProduct.destination=exchangeProduct
# 具体分组 对应 MQ 是 队列名称 并且持久化队列 inputProduct 自定义
spring.cloud.stream.bindings.inputProduct.group=groupProduct999
#开启消费者分区功能
spring.cloud.stream.bindings.inputProduct.consumer.partitioned=true
#指定了当前消费者的总实例数量
spring.cloud.stream.instanceCount=2
#设置当前实例的索引号，从 0 开始
spring.cloud.stream.instanceIndex=0实例B
spring.application.name=stream-partition-receiverB
server.port=9071
#设置服务注册中心地址，指向另一个注册中心
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=/
# 对应 MQ 是 exchange 和消息发送者的 交换器是同一个
spring.cloud.stream.bindings.inputProduct.destination=exchangeProduct
# 具体分组 对应 MQ 是 队列名称 并且持久化队列 inputProduct 自定义
spring.cloud.stream.bindings.inputProduct.group=groupProduct999
#开启消费者分区功能
spring.cloud.stream.bindings.inputProduct.consumer.partitioned=true
#指定了当前消费者的总实例数量
spring.cloud.stream.instanceCount=2
#设置当前实例的索引号，从 1 开始
spring.cloud.stream.instanceIndex=1
标签：Stream,stream,Spring,rabbitmq,Cloud,spring,input,bindings,cloud 来源： https://blog.csdn.net/wu_foot/article/details/117202814