Spring Cloud Stream是一個構(gòu)建消息驅(qū)動微服務(wù)的框架，抽象了MQ的使用方式, 提供統(tǒng)一的API操作。Spring Cloud Stream通過Binder(綁定器)、inputs/outputs Channel完成應(yīng)用程序和MQ的解耦。

• Binder
  負(fù)責(zé)綁定應(yīng)用程序和MQ中間件,即指定應(yīng)用程序是和KafKa交互還是和RabbitMQ交互或者和其他的MQ中間件交互

• inputs/outputs Channel
  inputs/outputs Channel抽象發(fā)布訂閱消息的方式，即無論是什么類型的MQ應(yīng)用程序都通過統(tǒng)一的方式發(fā)布訂閱消息

我們已經(jīng)搭建好了Kafka（參考Kafka單節(jié)點安裝），本文主要介紹一下Spring Cloud Stream與Kafka進(jìn)行集成實現(xiàn)消息的生產(chǎn)及消費。

項目創(chuàng)建

首先需要創(chuàng)建一個SpringBoot項目，命名為：spring-integration-kafka，在配置文件中導(dǎo)入相關(guān)的依賴。
項目情況為：

• 構(gòu)建工具：Gradle
• SpringBoot版本：2.7.5
• SpringBoot依賴管理版本：1.0.15.RELEASE
• SpringCloud依賴管理版本：2021.0.5

項目依賴

配置文件build.gradle.kts的關(guān)鍵配置項如下：

plugins {
    id("org.springframework.boot") version "2.7.5"
    id("io.spring.dependency-management") version "1.0.15.RELEASE"
}

apply(plugin = "org.springframework.boot")
apply(plugin = "io.spring.dependency-management")
apply(plugin = "java")


extra["springCloudVersion"] = "2021.0.5"

dependencyManagement {
    imports {
        mavenBom("org.springframework.cloud:spring-cloud-dependencies:${property("springCloudVersion")}")
    }
}

dependencies {
	implementation("org.springframework.boot:spring-boot-starter-web")
	implementation("org.springframework.boot:spring-boot-starter-actuator")
	
    implementation("org.springframework.cloud:spring-cloud-starter-bootstrap")
    implementation("org.springframework.cloud:spring-cloud-starter-stream-kafka")
    
    implementation("io.springfox:springfox-boot-starter:3.0.0")
    implementation("com.github.xiaoymin:swagger-bootstrap-ui:1.9.6")
}

集成配置

定義配置文件application.yml，配置文件中主要配置Kafka的地址、以及Spring Colud Stream的Binder和inputs/outputs Channel，其中：kafkaChannel1用于向Kafka發(fā)送消息；kafkaChannel2用于消費Kafka的消息。

spring:
  kafka:
    bootstrap-servers: wux-labs-vm:9092  # 定義Kafka的地址
    producer:
      acks: 1
  cloud:
    stream:
      binders:
        kafkaBiner1:   # 定義一個Binder，名稱隨意
          type: kafka   # Binder的類型是 kafka
          environment:
            spring:
              kafka: ${spring.kafka}  # Binder的配置使用前面配置的Kafka的信息
      default-binder: kafkaBiner1    # 默認(rèn)Binder，是前面配置的Binder的名稱
      bindings:
        kafkaChannel1:      # 定義一個（作為outputs Channel）通道，名稱隨意，在代碼中使用該通道名稱即可
          binder: kafkaBiner1 # 使用kafkaBiner1
          destination: KafkaFirstTopic # 定義目標(biāo)Topic的名稱
        kafkaChannel2:       # 定義一個（作為inputs Channel）通道，名稱隨意，在代碼中使用該通道名稱即可
          binder: kafkaBiner1 # 使用kafkaBiner1
          destination: KafkaFirstTopic # 定義目標(biāo)Topic的名稱
          group: group0   # 作為消息的消費方，需要指定group

集成生產(chǎn)者

下面開發(fā)一個生產(chǎn)者，發(fā)送消息需要通過outputs Channel進(jìn)行，使用kafkaChannel1發(fā)送消息到Kafka。

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Api("生產(chǎn)者接口")
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private StreamBridge bridge;

    @ApiOperation("向Kafka發(fā)送數(shù)據(jù)")
    @PostMapping("/kafka")
    public String sendToKafka(String message) {
        boolean status = bridge.send("kafkaChannel1", message);
        return "發(fā)送消息：" + message + "=====>" + status;
    }
}

集成消費者

消費消息需要通過inputs Channel進(jìn)行，定義一個Processor，指定訂閱通道為kafkaChannel2，這個通道被用于進(jìn)行消息消費，需要定義group。

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;

@Component
public interface ConsumerProcessor {
    @Input("kafkaChannel2")
    SubscribableChannel subscribableChannel();
}

啟用通道并監(jiān)聽。

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;

@EnableBinding(ConsumerProcessor.class)
public class ConsumerProcessorImpl {
    @StreamListener("kafkaChannel2")
    public void kafkaStreamListener(Object message) {
        System.out.println("接收到Kafka消息：" + new String((byte[]) message));
    }
}

集成驗證

生產(chǎn)者驗證

首先啟動一個Kafka自帶的消費者，監(jiān)聽KafkaFirstTopic。

接下來啟動SpringBoot項目并發(fā)送消息。在消費者那里可以看到接收到的消息。

消費者驗證

前面消息已經(jīng)發(fā)送到了Kafka的Topic了，可以看到控制臺直接打印出了監(jiān)聽到的消息。

至此，Spring Cloud Stream集成Kafka完成。文章來源地址http://www.zghlxwxcb.cn/news/detail-643671.html

到了這里，關(guān)于Spring Cloud Stream集成Kafka的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！