Spring Cloud构建微服务架构:消息驱动的微服务(消费分区)【Dalston版】
时间:2022-05-05
本文章向大家介绍Spring Cloud构建微服务架构:消息驱动的微服务(消费分区)【Dalston版】,主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
通过上一篇《消息驱动的微服务(消费组)》的学习,我们已经能够在多实例环境下,保证同一消息只被一个消费者实例进行接收和处理。但是,对于一些特殊场景,除了要保证单一实例消费之外,还希望那些具备相同特征的消息都能够被同一个实例进行消费。这时候我们就需要对消息进行分区处理。
使用消息分区
在Spring Cloud Stream中实现消息分区非常简单,我们可以根据消费组示例做一些配置修改就能实现,具体如下:
- 在消费者应用
SinkReceiver
中,我们对配置文件做一些修改,具体如下:
spring.cloud.stream.bindings.input.group=Service-A
spring.cloud.stream.bindings.input.destination=greetings
spring.cloud.stream.bindings.input.consumer.partitioned=true
spring.cloud.stream.instanceCount=2
spring.cloud.stream.instanceIndex=0
从上面的配置中,我们可以看到增加了这三个参数:
-
spring.cloud.stream.bindings.input.consumer.partitioned
:通过该参数开启消费者分区功能; -
spring.cloud.stream.instanceCount
:该参数指定了当前消费者的总实例数量; -
spring.cloud.stream.instanceIndex
:该参数设置当前实例的索引号,从0开始,最大值为spring.cloud.stream.instanceCount
参数 - 1。我们试验的时候需要启动多个实例,可以通过运行参数来为不同实例设置不同的索引值。
- 在生产者应用
SinkSender
中,我们对配置文件也做一些修改,具体如下:
spring.cloud.stream.bindings.output.destination=greetings
spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload
spring.cloud.stream.bindings.output.producer.partitionCount=2
从上面的配置中,我们可以看到增加了这两个参数:
-
spring.cloud.stream.bindings.output.producer.partitionKeyExpression
:通过该参数指定了分区键的表达式规则,我们可以根据实际的输出消息规则来配置SpEL来生成合适的分区键; -
spring.cloud.stream.bindings.output.producer.partitionCount
:该参数指定了消息分区的数量。
到这里消息分区配置就完成了,我们可以再次启动这两个应用,同时消费者启动多个,但需要注意的是要为消费者指定不同的实例索引号,这样当同一个消息被发给消费组时,我们可以发现只有一个消费实例在接收和处理这些相同的消息。
博客原文:http://blog.didispace.com/spring-cloud-starter-dalston-7-4/
但对依赖的Spring Boot和Spring Cloud版本做了升级。
- java教程
- Java快速入门
- Java 开发环境配置
- Java基本语法
- Java 对象和类
- Java 基本数据类型
- Java 变量类型
- Java 修饰符
- Java 运算符
- Java 循环结构
- Java 分支结构
- Java Number类
- Java Character类
- Java String类
- Java StringBuffer和StringBuilder类
- Java 数组
- Java 日期时间
- Java 正则表达式
- Java 方法
- Java 流(Stream)、文件(File)和IO
- Java 异常处理
- Java 继承
- Java 重写(Override)与重载(Overload)
- Java 多态
- Java 抽象类
- Java 封装
- Java 接口
- Java 包(package)
- Java 数据结构
- Java 集合框架
- Java 泛型
- Java 序列化
- Java 网络编程
- Java 发送邮件
- Java 多线程编程
- Java Applet基础
- Java 文档注释
- Scaling与Normalization的区别
- Android 音乐APP(一)扫描本地音乐
- Semaphore:如何快速实现一个限流器?
- 延迟初始化Spring Bean:延迟初始化的Bean会影响依赖注入吗?
- Android 音乐APP(二)启动白屏优化、定位当前播放歌曲
- CompletableFuture:异步编程没那么难
- 你知道Oracle数据库除了SGA和PGA,还有MGA么?
- .NET Standard 类库的使用技巧
- Makefile实例:批量编译多个目标
- GitHub如何回滚代码1.git log 查看commit hash值2.执行git reset --hard xxxx3.git push -f
- java获取程序执行时间
- Redis数据增多了,是该加内存还是加实例?
- 针对mysql delete删除表数据后占用空间不变小的问题
- 报错信息:(VI_1): ip address associated with VRID 80 not present in MASTER advertt : 192.168.1.8
- 如何保证集合是线程安全的? ConcurrentHashMap如何实现高效地线程安全?