SpringCloud?Stream?快速入門實例教程

更新時間：2025年11月26日 10:34:29 作者：Dajiaonew

本文介紹了Spring?Cloud?Stream（SCS）組件在分布式系統(tǒng)中的作用,以及如何集成到Spring?Boot項目中,通過SCS,可以統(tǒng)一管理不同的消息隊列中間件,并提供了更高級的抽象,感興趣的朋友跟隨小編一起看看吧

1.SCS 組件的出現(xiàn)的背景和作用

在分布式系統(tǒng)中，可能使用到的消息隊列讓人眼花繚亂，可能有使用（RabbitMq RroketMQ Kafka....），他們提供的客戶端各不相同，使用的方式也讓人眼花繚亂，此時就需要一個能夠統(tǒng)一消息隊列的客戶端，通過更高級的抽象來實現(xiàn)更通用和更簡單的集成不同的消息隊列中間件，此時也就誕生了這個SCS 組件

2.SCS 集成srping Boot項目

我們在這個演示項目中所使用的Spring Boot版本為 2.7.18、SpringCloud Alibaba版本為 2021.0.6.0

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>2021.0.6.0</version>
                <type>pom</type>
                <scope> import</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>2.7.18</version>
                <type>pom</type>
                <scope> import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

使用的SCS 組件版本為 3.2.10

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-kafka</artifactId>
            <version>3.2.10</version>
        </dependency>

3.Yml 配置

scs 的使用難點主要就是在yml 的配置上，配置完成使用很方便

spring:
  cloud:
    function:
      definition: myTaskConsumer;ackConsumer #你注冊的 Consumer 方法名 或者 Function 方法名 中間使用 ;分割 （生產(chǎn)者一般是動態(tài)發(fā)送消息 不需要注冊）
    stream:
      binders:
        kafka-binder-1: # 綁定器名稱
          type: kafka  # 消息隊列的類型類型
          environment: #綁定器環(huán)境配置
            spring:
              kafka:
                bootstrap-servers: 172.22.134.135:9092 # kafka地址 可以設(shè)置多個
                properties:
                  security.protocol: PLAINTEXT # kafka協(xié)議
        #rabbit-binde-1r:
         # type: rabbit
          # ... rabbitmq配置
      # 全局生產(chǎn)者可靠性配置（推薦）
      binder:
        producer-properties:
          acks: all                # ?? 生產(chǎn)者 ACK = all 所有副本同步完成才ack;ACK=1寫入leader副本返回ack;ACK=0 生產(chǎn)者發(fā)送消息立馬ack
          retries: 100          # 最大重試 當(dāng)發(fā)送失敗時（如網(wǎng)絡(luò)抖動、Leader 切換），Producer 自動重試的最大次數(shù)。
          retry.backoff.ms: 1000 #每次重試之間的等待時間（毫秒）。
          enable-idempotence: true      # 冪等生產(chǎn)者（防重復(fù)）
      bindings:
        myTaskConsumer-in-0: #命名規(guī)則 ${方法名}-${消費者:in/生產(chǎn)者:out}-${數(shù)字:不能與其他相同}
          destination: test-kraft # topic
          group: my-consumer-group #消費者組
          binder: kafka-binder-1 # 綁定器 <--上面配置的綁定名稱
          consumer: # 消費者配置
            autoStartup: true # 是否自動啟動
            concurrency: 1  #啟動消費者實例數(shù)  （同屬于一個消費者組）
        myTaskProducer-out-0:
          destination: test-kraft # topic
          binder: kafka-binder-1 # 綁定器 <--上面配置的綁定名稱
          producer: # 生產(chǎn)者配置
              partitionCount: 1 # 應(yīng)與目標(biāo) Topic 的實際分區(qū)數(shù)一致。
                                # - 若小于實際分區(qū)數(shù)：僅使用部分分區(qū)，浪費并行能力；
                                # - 若大于實際分區(qū)數(shù)：發(fā)送時會因訪問不存在的分區(qū)而失?。?
              #使用消息頭中的 headers的 partitionKey 作為 key進(jìn)行分區(qū)
              partition-key-expression: headers.partitionKey # 分區(qū)鍵(分區(qū)規(guī)則根據(jù)key進(jìn)行hash落到分區(qū) 有助于落到指定分區(qū)順序消費)

4.SpringCloud Stream 3.X新特性函數(shù)編程

4.1.編寫消費者

Mesage 的包別導(dǎo)錯

import org.springframework.messaging.Message;

@Configuration
public class kafkaConsumer
{
	@Bean
	public Consumer<Message<String>> myTaskConsumer ()
	{
		System.out.println ("[初始化] myTaskConsumer Bean 已創(chuàng)建");
		return message -> System.out.println ("[myTaskConsumer] 收到消息: " + message.getPayload ());
	}
}

4.2.編寫動態(tài)生產(chǎn)者

@RestController
public class SendController
{
	@Autowired
	StreamBridge streamBridge;
	@GetMapping ("/sendMyTaskProducer/{msg}")
	public String send (@PathVariable ("msg") String msg)
	{
		//構(gòu)建消息
		Message<String> message = MessageBuilder.withPayload (msg)
				.setHeader ("partitionKey", msg) // 添加分區(qū)鍵partitionKey 作為分區(qū)鍵
				.build ();
		//參數(shù)1為發(fā)送的通道名稱（在yml中配置），參數(shù)2為消息
		boolean myTaskProducer = streamBridge.send ("myTaskProducer-out-0", message);
		System.out.println ("發(fā)送結(jié)果：" + myTaskProducer);
		return "發(fā)送結(jié)果：" + myTaskProducer;
	}
}

5.進(jìn)行測試

訪問發(fā)送消息的接口，發(fā)送成功，并且消費者進(jìn)行了消費

6.進(jìn)行消費者手動ACK

消費者手動ACK 比自動ACK 要安全得多，默認(rèn)scs 是實行自動ack，自動ack只要消息被投遞到消費者，不論是否消費成功或者失敗，都會被視為消費成功

6.1yml 配置

        #========================================消費者ACK Kafka 專屬配置========================================
        #演示消費者ACK機(jī)制
        ackConsumer-in-0: #命名規(guī)則 ${方法名}-${消費者:in/生產(chǎn)者:out}-${數(shù)字:不能與其他相同}
          destination: topicOne # topic
          group: ack-consumer-group #消費者組 (修改為獨立的消費者組，避免與myTaskConsumer沖突)
          binder: kafka-binder-1 # 綁定器 <--上面配置的綁定名稱
          consumer: # 消費者配置
            autoStartup: true # 是否自動啟動
            concurrency: 1  #啟動消費者實例數(shù)  （同屬于一個消費者組）
        ackProducer-out-0:
          destination: topicOne # topic
          binder: kafka-binder-1 # 綁定器 <--上面配置的綁定名稱
          producer: # 生產(chǎn)者配置
            partitionCount: 1 # 應(yīng)與目標(biāo) Topic 的實際分區(qū)數(shù)一致。
              # - 若小于實際分區(qū)數(shù)：僅使用部分分區(qū)，浪費并行能力；
            # - 若大于實際分區(qū)數(shù)：發(fā)送時會因訪問不存在的分區(qū)而失敗！
            #使用消息頭中的 headers的 partitionKey 作為 key進(jìn)行分區(qū)
            partition-key-expression: headers.partitionKey # 分區(qū)鍵(分區(qū)規(guī)則根據(jù)key進(jìn)行hash落到分區(qū) 有助于落到指定分區(qū)順序消費)
      # Kafka 專屬配置
      kafka:
        bindings:
          ackConsumer-in-0: # ??指定哪個消費者使用ACK
            consumer:
              ack-mode: MANUAL  # ?? 關(guān)鍵！手動 ACK 模式
                                #RECORD	每條消息處理完自動提交 offset（默認(rèn)）	簡單場景
                                #BATCH	批量提交（一批 poll 的消息處理完后提交）	默認(rèn)行為（等價于 auto-commit=true）
                                #TIME	定時提交	較少用
                                #COUNT	每 N 條提交一次	較少用
                                #MANUAL	手動調(diào)用 acknowledge() 才提交	? 需要精確控制（推薦）
                                #MANUAL_IMMEDIATE	手動調(diào)用立即提交（不等批次）	高可靠性要求

6.2編寫消費者

@Bean
	public Consumer<Message<String>> ackConsumer(){
		System.out.println ("[初始化] ackConsumer Bean 已創(chuàng)建");
		return message  -> {
			System.out.println ("[ackConsumer] ========== 開始處理消息 ==========");
			System.out.println ("[ackConsumer] 消息內(nèi)容: " + message.getPayload());
			System.out.println ("[ackConsumer] 消息Headers: " + message.getHeaders());
			//獲取Acknowledgment
			Acknowledgment ack = message.getHeaders ()
					.get (KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
			if (ack != null) {
				//進(jìn)行手動ack
				ack.acknowledge ();
				System.out.println ("[ackConsumer] ? 已手動確認(rèn)消息");
			} else {
				System.out.println ("[ackConsumer] ?? 警告: Acknowledgment為null,無法手動確認(rèn)");
			}
			System.out.println ("[ackConsumer] ========== 消息處理完成 ==========\n");
		};
	}

6.3編寫生產(chǎn)者

@GetMapping ("/sendAckProducer/{msg}")
	public String send2 (@PathVariable ("msg") String msg)
	{
		//構(gòu)建消息
		Message<String> message = MessageBuilder.withPayload (msg)
				.setHeader ("partitionKey", msg) // 添加分區(qū)鍵partitionKey 作為分區(qū)鍵
				.build ();
		//參數(shù)1為發(fā)送的通道名稱（在yml中配置），參數(shù)2為消息
		boolean ackProducer = streamBridge.send ("ackProducer-out-0", message);
		System.out.println ("發(fā)送結(jié)果：" + ackProducer);
		return "發(fā)送結(jié)果：" + ackProducer;
	}