SpringBoot+Nacos+Kafka微服務流編排的簡單實現

Posted on by

前言

最近一直在做微服務開發，涉及瞭一些數據處理模塊的開發，每個處理業務都會開發獨立的微服務，便於後面拓展和流編排，學習瞭SpringCloud Data Flow等框架，感覺這個框架對於我們來說太重瞭，維護起來也比較麻煩，於是根據流編排的思想，基於我們目前的技術棧實現簡單的流編排功能。

簡單的說，我們希望自己的流編排就是微服務可插拔，微服務數據入口及輸出可不停機修改。

準備工作

Nacos安裝及使用入門

自己學習的話推薦使用docker安裝，命令如下

  • 拉取鏡像 docker pull nacos/nacos-server
  • 創建服務 docker run –env MODE=standalone –name nacos -d -p 8848:8848 nacos/nacos-server

然後在瀏覽器輸入 ip:8848/nacos 賬號nacos 密碼nacos

docker能夠幫助我們快速安裝服務，減少再環境準備花的時間

準備三個SpringBoot服務，引入Nacos及Kafka

<parent>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-parent</artifactId>
   <version>2.1.0.RELEASE</version>
</parent>

<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
</dependency>

<dependency>
   <groupId>com.alibaba.boot</groupId>
   <artifactId>nacos-config-spring-boot-starter</artifactId>
   <version>0.2.1</version>
</dependency>

配置文件 

spring:
  kafka:
    bootstrap-servers: kafka-server:9092
    producer:
      acks: all
    consumer:
      group-id: node1-group #三個服務分別為node1 node2 node3
      enable-auto-commit: false
# 部署的nacos服務
nacos:
  config:
    server-addr: nacos-server:8848

建議配置本機host 就可以填寫xxx-server 不用填寫服務ip

業務解讀

我們現在需要對三個服務進行編排，保障每個服務可以插拔，也可以調整服務的位子示意圖如下：

  • node1服務監聽前置服務發送的數據流，輸入的topic為前置數據服務輸出topic
  • node2監聽node1處理後的數據，所以node2監聽的topic為node1輸出的topic,node3同理，最終node3處理完成後將數據發送到數據流終點
  • 我們現在要調整流程 移除node2-server,我們隻需要把node1-sink改變成node2-sink即可，這樣我們這幾個服務就可以靈活的嵌入的不同項目的數據流處理業務中，做到即插即用（當然，數據格式這些業務層面的都是需要約定好的）
  • 動態可調還可以保證服務某一節點出現問題時候，即時改變數據流向，比如發送到數暫存服務，避免Kafka中積累太多數據，吞吐不平衡

Nacos配置

創建配置

通常流編排裡面每個服務都有一個輸入及輸出，分別為input及sink，所以每個服務我們需要配置兩個topic，分別是input-topic output-topic，我們就在nacos裡面添加輸入輸出配置

nacos配置項需要配置groupId,dataId,通常我們用服務名稱作為groupId,配置項的名稱作為dataId,如node1-server服務有一個input配置項，配置如下:

完成其中一個服務的配置，其它服務參考下圖配置即可

讀取配置

@Configuration
@NacosPropertySource(dataId = "input", groupId = "node1-server", autoRefreshed = true)
// autoRefreshed=true指的是nacos中配置發生改變後會刷新，false代表隻會使用服務啟動時候讀取到的值
@NacosPropertySource(dataId = "sink", groupId = "node1-server", autoRefreshed = true)
public class NacosConfig {

    @NacosValue(value = "${input:}", autoRefreshed = true)
    private String input;

    @NacosValue(value = "${sink:}", autoRefreshed = true)
    private String sink;
    
    public String getInput() {
        return input;
    }
    
    public String getSink() {
        return sink;
    }
}

監聽配置改變

服務的輸入需要在服務啟動時候創建消費者，在topic發生改變時候重新創建消費者，移除舊topic的消費者，輸出是業務驅動的，無需監聽改變，在每次發送時候讀取到的都是最新配置的topic，因為在上面的配置類中autoRefreshed = true，這個隻會刷新nacosConfig中的配置值，服務需要知道配置改變去驅動消費的創建業務，需要創建nacos配置監聽

/**
 * 監聽Nacos配置改變，創建消費者，更新消費
 */
@Component
public class ConsumerManager {

    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private boolean enableAutoCommit;
    
    @Value("${spring.kafka.consumer.group-id}")
    private boolean groupId;
    
    @Autowired
    private NacosConfig nacosConfig;
    
    @Autowired
    private KafkaTemplate kafkaTemplate;
    
    // 用於存放當前消費者使用的topic
    private String topic;
    
    // 用於執行消費者線程
    private ExecutorService executorService;
    
    /**
     * 監聽input
     */
    @NacosConfigListener(dataId = "node1-server", groupId = "input")
    public void inputListener(String input) {
        // 這個監聽觸發的時候 實際NacosConfig中input的值已經是最新的值瞭 我們隻是需要這個監聽觸發我們更新消費者的業務
        String inputTopic = nacosConfig.getInput();
        // 我使用nacosConfig中讀取的原因是因為監聽到內容是input=xxxx而不是xxxx,如果使用需要自己截取一下,nacosConfig中的內容框架會處理好，大傢看一下第一張圖的配置內容就明白瞭
        // 先檢查當前局部變量topic是否有值,有值代表是更新消費者，沒有值隻需要創建即可
        if(topic != null) {
            // 停止舊的消費者線程
            executorService.shutdownNow();
            executorService == null;
        }
        // 根據為新的topic創建消費者
        topic = inputTopic;
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(topic + "-pool-%d").build();
        executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(2), threadFactory);
        // 執行消費業務
        executorService.execute(() -> consumer(topic));
    }
    
    /**
     * 創建消費者
     */
    public void consumer(String topic) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        properties.put("enable.auto.commit", enableAutoCommit);
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", groupId);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(topic));
        try {
            while (!Thread.currentThread().isInterrupted()) {
                Duration duration = Duration.ofSeconds(1L);
                ConsumerRecords<String, String> records = consumer.poll(duration);
                for (ConsumerRecord<String, String> record : records) {
                    String message = record.value();
                    // 執行數據處理業務 省略業務實現
                    String handleMessage =  handle(message);
                    // 處理完成後發送到下一個節點
                    kafkaTemplate.send(nacosConfig.getSink(), handleMessage);
                }
            }
            consumer.commitAsync();
        }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
}

總結

流編排的思路整體來說就是數據流方向可調，我們以此為需求，根據一些主流框架提供的api實現自己的動態調整方案，可以幫助自己更好的理解流編碼思想及原理，在實際業務中，還有許多業務問題需要去突破，我們這樣處理更多是因為服務可插拔，便於流處理微服務在項目靈活搭配，因為我現在工作是在傳統公司，由於一些原因很難去推動新框架的使用，經常會用一些現有技術棧組合搞一些sao操作，供大傢參考，希望大傢多多指教。

到此這篇關於SpringBoot+Nacos+Kafka微服務流編排的簡單實現的文章就介紹到這瞭,更多相關SpringBoot+Nacos+Kafka流編排內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：