文档章节

Spring Cloud构建微服务架构:消息驱动的微服务(消费组)【Dalston版】

程序猿DD
 程序猿DD
发布于 03/06 07:26
字数 737
阅读 373
收藏 3

通过之前的《消息驱动的微服务(入门)》一文,相信很多朋友已经对Spring Cloud Stream有了一个初步的认识。但是,对于《消息驱动的微服务(核心概念)》一文中提到的一些核心概念可能还有些迷糊,下面我们将详细的来学习一下这些概念。本文我们就来学习和使用一下“消费组”这一概念。

使用消费组实现消息消费的负载均衡

通常在生产环境,我们的每个服务都不会以单节点的方式运行在生产环境,当同一个服务启动多个实例的时候,这些实例都会绑定到同一个消息通道的目标主题(Topic)上。

默认情况下,当生产者发出一条消息到绑定通道上,这条消息会产生多个副本被每个消费者实例接收和处理,但是有些业务场景之下,我们希望生产者产生的消息只被其中一个实例消费,这个时候我们需要为这些消费者设置消费组来实现这样的功能,实现的方式非常简单,我们只需要在服务消费者端设置spring.cloud.stream.bindings.input.group属性即可,比如我们可以这样实现:

  • 先创建一个消费者应用SinkReceiver,实现了greetings主题上的输入通道绑定,它的实现如下:
@EnableBinding(value = {Sink.class})
public class SinkReceiver {

    private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class);

    @StreamListener(Sink.INPUT)
    public void receive(User user) {
        logger.info("Received: " + user);
    }
}

  • 为了将SinkReceiver的输入通道目标设置为greetings主题,以及将该服务的实例设置为同一个消费组,做如下设置:
spring.cloud.stream.bindings.input.group=Service-A
spring.cloud.stream.bindings.input.destination=greetings

通过spring.cloud.stream.bindings.input.group属性指定了该应用实例都属于Service-A消费组,而spring.cloud.stream.bindings.input.destination属性则指定了输入通道对应的主题名。

  • 完成了消息消费者之后,我们再来实现一个消息生产者应用SinkSender,具体如下:
@EnableBinding(value = {Source.class})
public class SinkSender {

    private static Logger logger = LoggerFactory.getLogger(SinkSender.class);

    @Bean
    @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000"))
    public MessageSource<String> timerMessageSource() {
        return () -> new GenericMessage<>("{\"name\":\"didi\", \"age\":30}");
    }

}
  • 为消息生产者SinkSender做一些设置,让它的输出通道绑定目标也指向greetings主题,具体如下:
spring.cloud.stream.bindings.output.destination=greetings

到这里,对于消费分组的示例就已经完成了。分别运行上面实现的生产者与消费者,其中消费者我们启动多个实例。通过控制台,我们可以发现每个生产者发出的消息,会被启动的消费者以轮询的方式进行接收和输出。

以下专题教程也许您会有兴趣

本文内容部分节选自我的《Spring Cloud微服务实战》,稍作改变和内容升级

© 著作权归作者所有

共有 人打赏支持
程序猿DD
粉丝 348
博文 59
码字总数 80114
作品 0
闵行
加载中

评论(1)

黑暗骑士C
黑暗骑士C
Java主流技术分享:SpringBoot、SpringCloud、Docker、dubbo、redis、多线程高并发、数据库性能调优等等。https://charllee.github.io/2018/02/27/Java%E6%9C%80%E6%96%B0%E5%AD%A6%E4%B9%A0%E8%B5%84%E6%96%99-2018/
程序猿DD/SpringCloud-Learning

Spring Cloud教程 本项目内容为Spring Cloud教程的程序样例。如您觉得该项目对您有用,欢迎点击右上方的Star按钮,给予支持!! 我的博客:http://blog.didispace.com 我的小密圈(深度交流与...

程序猿DD
2016/11/03
0
0
SpringCloud_学习_00_资源帖

一、官方资料 1.springcloud中文网 2.Spring Cloud 官方中文文档(dalston) 3.spring cloud Quick Start 二、github 精选 1.翟永超-《Spring Cloud构建微服务架构》系列博文Dalston版 2.方志...

rayner
04/13
0
0
微服务选择Spring Cloud还是Dubbo?

点击关注 异步图书,置顶公众号 每天与你分享 IT好书 技术干货 职场知识 参与文末话题讨论,每日赠送异步图书。 ——异步小编 在阿里巴巴的生态中,微服务逐渐成为主要的服务形态,伴随着容器...

异步社区
04/27
0
0
Spring Cloud构建微服务架构:消息驱动的微服务(入门)【Dalston版】

之前在写Spring Boot基础教程的时候写过一篇《Spring Boot中使用RabbitMQ》。在该文中,我们通过简单的配置和注解就能实现向RabbitMQ中生产和消费消息。实际上我们使用的对RabbitMQ的starter...

翟永超
2017/12/18
0
0
Spring Cloud入门教程(十):消息总线(Bus)

Spring Cloud入门教程系列: Spring Cloud入门教程(一):服务治理(Eureka) Spring Cloud入门教程(二):客户端负载均衡(Ribbon) Spring Cloud入门教程(三):声明式服务调用(Feign) Spring Cl...

CD826
05/22
0
0

没有更多内容

加载失败,请刷新页面

加载更多

GO 数组相关操作

package mainimport("fmt""math/rand""time")func main() {//数组的几种定义方式var arr1 [3]int = [3]int{1,2,3}var arr2 = [3]int{4,5,6}arr3 := [3]string{"h", "w", ......

汤汤圆圆
43分钟前
1
0
JAVA 中interrupt、interrupted和isInterrupted的区别

首先,我们说明下三个方法的功能 interrupt() 向当前调用者线程发出中断信号 isinterrupted() 查看当前中断信号是true还是false interrupted() 是静态方法,查看返回当前中断信号并将中断信号...

我爱春天的毛毛雨
47分钟前
1
0
Coding and Paper Letter(二十二)

资源整理。 1 Coding: 1.开源项目openeo api。oponEO开发了一个开放的API,以简单统一的方式将R,python和javascript客户端连接到对地观测大数据云平台的后台。 此存储库包含此API,即oponE...

胖胖雕
今天
1
0
RxJS的另外四种实现方式(三)——性能最高的库

接上篇 RxJS的另外四种实现方式(二)——代码最小的库(续) 代码最小的库rx4rx-lite虽然在性能测试中超过了callbag,但和most库较量的时候却落败了,于是我下载了most库,要解开most库性能...

一个灰
今天
6
0
马太效应

马太效应

yizhichao
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部