实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

这篇具有很好参考价值的文章主要介绍了实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

前言

前面的博文我们介绍并实战演示了Spring Cloud Stream整合rabbitmq,其中主要介绍了如何使用和配置完成消息中间件的集成。但是,在实际的生产环境中可能会用到多个消息中间件,又或者是由于业务改变需要更换消息中间件,在这些情况下我们的Spring Cloud Stream框架可以完全兼容多个消息中间件和多种消息中间件的替换。今天,我们就在一个项目中用Spring Cloud Stream 集成两个消息中间件kafka和rabbitmq。

实战要点

1、完美集成并兼容kafka和rabbitmq
2、增加消费组概念,直接保证消息唯一消费
3、增加重试机制,重试条件满足后自动加入死信
4、增加死信消费者,可以直接移植生产
5、消费者手动ack、offset
6、rabbitmq、kafka配置,保证消息不丢失

技术积累

Spring Cloud Stream简介

Spring Cloud Stream是用于构建微服务具有消息驱动能力的框架,应用程序通过inputs、outputs通道与binder进行交互,binder与消息中间件进行通信。

binder的作用是将消息中间件进行粘合,相当于对第三方中间件进行封装整合,让开发人员不用关心底层消息中间件如何运行。
实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

inputs是消息输入通道,类似于消息中间件的consumer消费者;outputs是消息输出通道,类似于消息中间件的producer生产者。应用程序收发消息不再直接调用消息中间件的接口或者逻辑代码,直接使用Spring Cloud Stream 的OUTPUT与INPUT通道进行处理。

可以通过binder绑定选用各种消息中间件,用binding进行中间件的相关参数配置,让应用程序达到灵活配置和切换消息中间件的目的。

集成kafka要点

1、修改server.properties文件,将#listeners=PLAINTEXT://:9092这一句注释放开,改为listeners=PLAINTEXT://kafka服务器ip:9092
如果此处不改SpringBoot在启动时会报错:
Error connecting to node devops-01:9092 (id: 0 rack: null)
2、kafka 2.8版本开始自带zk,建议使用2.8版本以上的版本不用安装zk
3、spring-boot-starter-paren与spring-cloud-starter-stream-kafk版本号一定要对应上,特别是springboot2之后的版本。如果没有特殊要求,需严格按照本文的版本号进行配置和实战
4、kafka本身、生产者、消费者保证消息不丢失,注意必须使用kafka HA配合修改配置

集成rabbitmq要点

1、rabbitmq比kafka的限制条件就少很多,基本上不用考虑spring版本号兼容
2、rabbimq本身、生产者、消费者保证消息不丢失,注意必须使用rabbitmq HA

实战演示

本次实战直接采用从0到1的策略进行演示,适合小白直接入手,可直接接入生产
本次实战MQ组件全部采用单机进行测试,生产环境请更换为HA

本次实战提供:
1、Kafka、Rabbitmq消息中间件信道注册
2、Kafka、Rabbitm消息中间件消息发送、接收消息监听、死信消息监听

Maven依赖版本号选择

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.12.RELEASE</version>
    <relativePath/>
</parent>
<properties>
    <java.version>1.8</java.version>
    <spring-cloud.version>Hoxton.SR10</spring-cloud.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-kafka</artifactId>
        <version>3.0.3.RELEASE</version>
    </dependency>
</dependencies>
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

Spring及MQ主要配置

server:
  port: 9999
spring:
  rabbitmq:
    host: 10.10.22.187
    port: 5672
    username: admin
    password: admin
    virtual-host: /
  kafka:
    bootstrap-servers: 10.10.22.174:9092
  cloud:
    stream:
      default-binder: myRabbit #默认绑定的mq
      binders: #stream框架粘接的mq
        myRabbit: #自定义个人mq名称
          type: rabbit
          environment:
            spring: ${spring.rabbitmq}
        myKafka:
          type: kafka
          environment:
            spring:
              cloud:
                stream:
                  kafka: ${spring.cloud.stream.kafka.binder}
      bindings: #stream绑定信道
        output_channel: #自定义发送信道名称
          destination: assExchange #目的地 交换机/主题
          content-type: application/json
          binder: myRabbit #粘接到的mq
        input_channel: #自定义接收信道
          destination: assExchange #目的地 交换机/主题
          content-type: application/json
          binder: myRabbit #粘接到的mq
          group: assGroup
          consumer:
            maxAttempts: 3 # 尝试消费该消息的最大次数(消息消费失败后,发布者会重新投递)。默认3
            backOffInitialInterval: 1000 # 重试消费消息的初始化间隔时间。默认1s,即第一次重试消费会在1s后进行
            backOffMultiplier: 2 # 相邻两次重试之间的间隔时间的倍数。默认2
            backOffMaxInterval: 10000 # 下一次尝试重试的最大时间间隔,默认为10000ms,即10s
        output_kafka_channel: #自定义发送信道名称
          destination: assTopic #目的地 交换机/主题
          content-type: text/plain
          binder: myKafka #粘接到的mq
          producer:
            partition-count: 2 #分区数目
        input_kafka_channel: #自定义接收信道
          destination: assTopic #目的地 交换机/主题
          content-type: text/plain
          binder: myKafka #粘接到的mq
          group: assGroup
          consumer:
            maxAttempts: 3 # 尝试消费该消息的最大次数(消息消费失败后,发布者会重新投递)。默认3
            backOffInitialInterval: 1000 # 重试消费消息的初始化间隔时间。默认1s,即第一次重试消费会在1s后进行
            backOffMultiplier: 2 # 相邻两次重试之间的间隔时间的倍数。默认2
            backOffMaxInterval: 10000 # 下一次尝试重试的最大时间间隔,默认为10000ms,即10s
      rabbit: #stream mq配置
        bindings:
          out_channel:
            producer:
              delivery-mode: persistent   #消息持久化 non-persistent
              useConfirmHeader: true #Future<Confirm>获取异常投递,与confirmAckChannel互斥
          input_channel:
            consumer:
              concurrency: 1 #消费者数量
              max-concurrency: 5 #最大消费者数量
              durable-subscription: true  #持久化队列
              recovery-interval: 3000  #3s 重连
              acknowledge-mode: MANUAL  #手动
              requeue-rejected: false #是否重新放入队列
              auto-bind-dlq: true #开启死信队列
              requeueRejected: true #异常放入死信
      kafka:
        binder:
          brokers: ${spring.kafka.bootstrap-servers}
          auto-add-partitions: true #自动分区
          auto-create-topics: true #自动创建主题
          replication-factor: 1 #两个副本
          min-partition-count: 1 #最小分区
        bindings:
          out_kafka_channel:
            producer:
              # 无限制重发不产生消息丢失
              retries: Integer.MAX_VALUE
              #acks =0:producer不等待broker的ack,broker一接收到还没有写入磁盘就已经返回,可靠性最低
              #acks =1:producer等待broker的ack,partition的leader刷盘成功后返回ack,如果在follower同步成功之前leader故障,那么将会丢失数据,可靠性中
              #acks = all 、 -1:producer等待broker的ack,partition的leader和follower全部落盘成功后才返回ack,可靠性高,但延迟时间长
              #可以设置的值为:all, -1, 0, 1
              acks: all
              min:
                insync:
                  replicas: 1 #感知副本数
          input_kafka_channel:
            consumer:
              concurrency: 1 #消费者数量
              max-concurrency: 5 #最大消费者数量
              recovery-interval: 3000  #3s 重连
              auto-rebalance-enabled: true  #主题分区消费者组成员自动平衡
              auto-commit-offset: false   #手动提交偏移量
              enable-dlq: true  # 开启 dlq队列
              dlq-name: assTopic.dlq
              deserializationExceptionHandler: sendToDlq #异常加入死信

基础信道

/**
 * MqChannel
 * @author senfel
 * @version 1.0
 * @date 2023/6/2 15:46
 */
public interface MqChannel {

    /**
     * 消息目的地
     * RabbitMQ中为交换机名称
     * kafka topic
     */
    String DESTINATION = "assExchange";
    String DESTINATIONBYGROUP = "assGroup";
    String DESTINATIONBYTOPIC = "assTopic";
    /**
     * 输出信道
     */
    String OUTPUT_CHANNEL = "output_channel";
    String OUTPUT_KAFKA_CHANNEL = "output_kafka_channel";
    /**
     * 输入信道
     */
    String INPUT_CHANNEL = "input_channel";
    String INPUT_KAFKA_CHANNEL = "input_kafka_channel";
    String INPUT_KAFKA_CHANNEL_ERROR = "assTopic.dlq";
    /**
     * 死信队列
     */
    String INPUT_CHANNEL_DLQ = "assExchange.assGroup.dlq";


    @Output(MqChannel.OUTPUT_CHANNEL)
    MessageChannel output();

    @Output(MqChannel.OUTPUT_KAFKA_CHANNEL)
    MessageChannel outputByKafka();

    @Input(MqChannel.INPUT_CHANNEL)
    SubscribableChannel input();

    @Input(MqChannel.INPUT_KAFKA_CHANNEL)
    SubscribableChannel inputByKafka();

    @Input(MqChannel.INPUT_KAFKA_CHANNEL_ERROR)
    SubscribableChannel inputByKafkaError();
}

绑定信道消息发送

提供绑定信道,增加rabbitmq、kafka发消息逻辑

1、启动类增加绑定mq注解@EnableBinding(MqChannel.class)

@SpringBootApplication
@EnableBinding(MqChannel.class)
public class TestDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(TestDemoApplication.class, args);
    }

}

2、增加发送消息接口

/**
 * TestMQService
 * @author senfel
 * @version 1.0
 * @date 2023/6/2 15:47
 */
public interface TestMQService {

    /**
     * rabbitmq发送消息
     */
    void send(String str);

    /**
     * kafka发送消息
     */
    void sendByKafka(String str);


}

3、实现发送消息接口

/**
 * TestMQServiceImpl
 * @author senfel
 * @version 1.0
 * @date 2023/6/2 15:49
 */
@Service
@Slf4j
public class TestMQServiceImpl implements TestMQService {

    @Resource
    private MqChannel mqChannel;

    @Override
    public void send(String str) {
        mqChannel.output().send(MessageBuilder.withPayload("rabbitmq测试:"+str).build());
    }

    @Override
    public void sendByKafka(String str) {
        mqChannel.outputByKafka().send(MessageBuilder.withPayload("kafka测试:"+str).build());
    }
}

4、提供接口层

/**
 * @author senfel
 * @version 1.0
 * @date 2023/6/2 17:27
 */
@RestController
public class TestController{
    @Resource
    private TestMQService testMQService;


    /**
     * testRabbitmq
     * @param str
     * @author senfel
     * @date 2023/6/8 11:27 
     * @return java.lang.String
     */
    @GetMapping("/test")
    public String testMq(String str){
        testMQService.send(str);
        return str;
    }

    /**
     * testKafka
     * @param str
     * @author senfel
     * @date 2023/6/8 11:27 
     * @return java.lang.String
     */
    @GetMapping("/testKafka")
    public String testKafka(String str){
        testMQService.sendByKafka(str);
        return str;
    }
}

集成兼容多mq演示

Rabbitmq演示

1、TestMQServiceImpl增加mq消息监听和私信监听

 /**
 * 接收消息监听
 * @param message 消息体
 * @param channel 信道
 * @param tag 标签
 * @author senfel
 * @date 2023/6/5 9:25
 * @return void
 */
@StreamListener(MqChannel.INPUT_CHANNEL)
public void process(String message,
                    @Header(AmqpHeaders.CHANNEL) Channel channel,
                    @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
    log.info("message : "+message);
    if(message.contains("9")){
        // 参数1为消息的tag  参数2为是否多条处理 参数3为是否重发
        //channel.basicNack(tag,false,false);
        System.err.println("--------------rabbitmq消费者消费异常--------------------------------------");
        System.err.println(message);
        throw new RuntimeException("消费异常");
    }else{
        System.err.println("--------------rabbitmq消费者--------------------------------------");
        System.err.println(message);
        channel.basicAck(tag,false);
    }

}



/**
 * 死信监听
 * @param message 消息体
 * @param channel 信道
 * @param tag 标签
 * @author senfel
 * @date 2023/6/5 14:30
 * @return void
 */
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(MqChannel.INPUT_CHANNEL_DLQ)
                , exchange = @Exchange(MqChannel.DESTINATION)
        ),
        concurrency = "1-5"
)
public void processByDlq(String message,
                         @Header(AmqpHeaders.CHANNEL) Channel channel,
                         @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {

    log.info("message : "+message);
    System.err.println("---------------rabbitmq死信消费者------------------------------------");
    System.err.println(message);
}

2、测试正常消息投递

实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

--------------rabbitmq消费者--------------------------------------
rabbitmq测试:777777777777777

3、测试异常消息投递,投递规则3次消费失败直接进入死信

实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

--------------rabbitmq消费者消费异常--------------------------------------
rabbitmq测试:7777777777777779
--------------rabbitmq消费者消费异常--------------------------------------
rabbitmq测试:7777777777777779
--------------rabbitmq消费者消费异常--------------------------------------
rabbitmq测试:7777777777777779
---------------rabbitmq死信消费者------------------------------------
rabbitmq测试:7777777777777779

Kafka演示

1、TestMQServiceImpl增加mq消息监听和私信监听

   /**
     * kafka消费者
     * @param message 消息体
     * @param acknowledgment ack
     * @param receivedTopic topic
     * @param groupId 消费者group
     * @author senfel
     * @date 2023/6/7 15:59
     * @return void
     */
    @StreamListener(MqChannel.INPUT_KAFKA_CHANNEL)
    public void processByKafka(String message,
                               @Header(value = KafkaHeaders.ACKNOWLEDGMENT,required = false) Acknowledgment acknowledgment,
                               @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic,
                               @Header(value = KafkaHeaders.GROUP_ID,required = false) String groupId,
                               @Header(value = KafkaHeaders.PARTITION_ID,required = false) String partitionId) throws Exception {
        System.err.println("-------进入kafka消费者---------------");
        System.err.println(message);
        System.err.println(receivedTopic);
        if(message.contains("9")){
            log.error("kafka消费异常:{}",message);
            System.err.println("kafka1消费异常"+message);
            throw new RuntimeException("kafka消费异常");
        }
        System.err.println("kafka接受的数据为"+message);
        acknowledgment.acknowledge();
    }



    /**
     * kafka死信消费
     * @param message 消息体
     * @param receivedTopic  topic
     * @author senfel
     * @date 2023/6/7 15:58
     * @return void
     */
   @KafkaListener(topics = {MqChannel.INPUT_KAFKA_CHANNEL_ERROR},
                    groupId = MqChannel.DESTINATIONBYGROUP)
    public void processByKafkaError(String message,
                                    @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) throws Exception {

        System.err.println("-------进入死信消费者---------------");
        System.err.println(message);
        System.err.println(receivedTopic);
        System.err.println("kafka死信接受的数据为"+message);
        System.err.println(message);
    }

2、测试正常消息投递

实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

-------进入kafka消费者---------------
kafka测试:7777777777777777
assTopic
kafka接受的数据为kafka测试:7777777777777777

3、测试异常消息投递,投递规则3次消费失败直接进入死信

实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

-------进入kafka消费者---------------
kafka测试:7777777777777779
assTopic
kafka1消费异常kafka测试:7777777777777779
-------进入kafka消费者---------------
kafka测试:7777777777777779
assTopic
kafka1消费异常kafka测试:7777777777777779
-------进入kafka消费者---------------
kafka测试:7777777777777779
assTopic
kafka1消费异常kafka测试:7777777777777779
-------进入死信消费者---------------
kafka测试:7777777777777779
assTopic.dlq
kafka死信接受的数据为kafka测试:7777777777777779
kafka测试:7777777777777779

写在最后

Spring Cloud Stream集成多消息中间件kafka、rabbitmq较为简单,直接省去了原生中间的的操作与处理,开发人员可以直接任意切换和混用多种消息中间件,大大增加架构的可用性与可移植性。本实战案例提供重试、私信、手动ack、消费者分组和负载等高可用方案,直接可接入生产使用。

⭐️路漫漫其修远兮,吾将上下而求索 🔍文章来源地址https://www.toymoban.com/news/detail-477700.html

到了这里,关于实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 实战:Spring Cloud Stream消息驱动框架整合rabbitMq

    相信很多同学都开发过WEB服务,在WEB服务的开发中一般是通过缓存、队列、读写分离、削峰填谷、限流降级等手段来提高服务性能和保证服务的正常投用。对于削峰填谷就不得不用到我们的MQ消息中间件,比如适用于大数据的kafka,性能较高支持事务活跃度高的rabbitmq等等,MQ的

    2024年02月08日
    浏览(45)
  • 【Spring Cloud】服务容错中间件Sentinel入门

    欢迎来到阿Q社区 https://bbs.csdn.net/topics/617897123 Sentinel(分布式系统的流量防卫兵)是阿里开源的一套用于 服务容错 的综合性解决方案。它以流量为切入点,从 流量控制、熔断降级、系统负载保护 等多个维度来保护服务的稳定性。 Sentinel 具有以下特征: 丰富的应用场景 :

    2024年04月15日
    浏览(41)
  • 【Spring Cloud】服务容错中间件Sentinel进阶——五大规则

    我们在上一篇文章中对 Sentinel 已经有了基本的了解,接下来,我们一起对它的进阶进行学习吧! 资源 所谓资源就是 Sentinel 要保护的东西,资源是 Sentinel 的关键概念。它可以是Java应用程序中的任何内容,可以是一个服务,也可以是一个方法,甚至可以是一段代码。 我们入门

    2024年04月26日
    浏览(38)
  • 使用Spring Boot集成中间件:基础篇

    在现代应用开发中,中间件在构建高效、可扩展的系统方面起着至关重要的作用。而Spring Boot作为一种快速开发框架,提供了丰富的集成中间件的能力,使得我们能够轻松地将各种中间件引入到我们的应用程序中。本文将重点介绍如何使用Spring Boot集成Redis中间件,并提供一个

    2024年01月25日
    浏览(44)
  • Spring Cloud【消息驱动(什么是Spring Cloud Stream、SpringCloud Stream核心概念、入门案例之消息消费者 )】(十一)

      目录 消息驱动_什么是Spring Cloud Stream 消息驱动_SpringCloud Stream核心概念

    2024年02月15日
    浏览(41)
  • 使用Spring Boot集成中间件:Elasticsearch基础->提高篇

    Elasticsearch是一个开源的分布式搜索和分析引擎,广泛用于构建实时的搜索和分析应用。在本篇博客中,我们将深入讲解如何使用Spring Boot集成Elasticsearch,实现数据的索引、搜索和分析。 在开始之前,确保已经完成以下准备工作: 安装并启动Elasticsearch集群 创建Elasticsearch索引

    2024年01月19日
    浏览(45)
  • 使用Spring Boot集成中间件:Kafka的高级使用案例讲解

    在实际应用中,Kafka作为一种强大的分布式消息系统,广泛应用于实时数据处理和消息传递。本文将通过一个全面的使用案例,详细介绍如何使用Spring Boot集成Kafka,并展示其在实际场景中的应用。 在开始之前,我们需要确保已经完成以下准备工作: 安装并启动Kafka集群 创建

    2024年02月01日
    浏览(53)
  • 【消息中间件MQ系列】Spring整合kafka并设置多套kafka配置

            圣诞节的到来,程序员不会收到圣诞老人的🎁,但可以自己满足一下自己,所以,趁着有时间,就记录一下这会儿撸了些什么代码吧!!!         因为业务原因,需要在系统内新增其他的kakfa配置使用,所以今天研究的是怎么在系统内整合多套kafka配置使用。

    2024年02月01日
    浏览(96)
  • Spring Cloud Stream集成Kafka

    Spring Cloud Stream是一个构建消息驱动微服务的框架,抽象了MQ的使用方式, 提供统一的API操作。Spring Cloud Stream通过Binder(绑定器)、inputs/outputs Channel完成应用程序和MQ的解耦。 Binder 负责绑定应用程序和MQ中间件,即指定应用程序是和KafKa交互还是和RabbitMQ交互或者和其他的MQ中间件交

    2024年02月13日
    浏览(48)
  • Spring Cloud Stream 4.0.4 rabbitmq 发送消息多function

    注意当多个消费者时,需要添加配置项:spring.cloud.function.definition 启动日志 交换机名称对应: spring.cloud.stream.bindings.demo-in-0.destination配置项的值 队列名称是交换机名称+分组名 http://localhost:8080/sendMsg?delay=10000name=zhangsan 问题总结 问题一 解决办法: 查看配置是否正确: spring

    2024年02月19日
    浏览(44)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包