您好,登錄后才能下訂單哦!
這篇文章主要介紹實(shí)現(xiàn)SpringBoot+Nacos+Kafka微服務(wù)流編排的示例分析,文中介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們一定要看完!
最近一直在做微服務(wù)開發(fā),涉及了一些數(shù)據(jù)處理模塊的開發(fā),每個(gè)處理業(yè)務(wù)都會(huì)開發(fā)獨(dú)立的微服務(wù),便于后面拓展和流編排,學(xué)習(xí)了SpringCloud Data Flow等框架,感覺這個(gè)框架對于我們來說太重了,維護(hù)起來也比較麻煩,于是根據(jù)流編排的思想,基于我們目前的技術(shù)棧實(shí)現(xiàn)簡單的流編排功能。
簡單的說,我們希望自己的流編排就是微服務(wù)可插拔,微服務(wù)數(shù)據(jù)入口及輸出可不停機(jī)修改。
自己學(xué)習(xí)的話推薦使用docker安裝,命令如下
拉取鏡像 docker pull nacos/nacos-server
創(chuàng)建服務(wù) docker run --env MODE=standalone --name nacos -d -p 8848:8848 nacos/nacos-server
然后在瀏覽器輸入 ip:8848/nacos 賬號nacos 密碼nacos
docker能夠幫助我們快速安裝服務(wù),減少再環(huán)境準(zhǔn)備花的時(shí)間
<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 #三個(gè)服務(wù)分別為node1 node2 node3 enable-auto-commit: false # 部署的nacos服務(wù) nacos: config: server-addr: nacos-server:8848
建議配置本機(jī)host 就可以填寫xxx-server 不用填寫服務(wù)ip
我們現(xiàn)在需要對三個(gè)服務(wù)進(jìn)行編排,保障每個(gè)服務(wù)可以插拔,也可以調(diào)整服務(wù)的位子示意圖如下:
node1服務(wù)監(jiān)聽前置服務(wù)發(fā)送的數(shù)據(jù)流,輸入的topic為前置數(shù)據(jù)服務(wù)輸出topic
node2監(jiān)聽node1處理后的數(shù)據(jù),所以node2監(jiān)聽的topic為node1輸出的topic,node3同理,最終node3處理完成后將數(shù)據(jù)發(fā)送到數(shù)據(jù)流終點(diǎn)
我們現(xiàn)在要調(diào)整流程 移除node2-server,我們只需要把node1-sink改變成node2-sink即可,這樣我們這幾個(gè)服務(wù)就可以靈活的嵌入的不同項(xiàng)目的數(shù)據(jù)流處理業(yè)務(wù)中,做到即插即用(當(dāng)然,數(shù)據(jù)格式這些業(yè)務(wù)層面的都是需要約定好的)
動(dòng)態(tài)可調(diào)還可以保證服務(wù)某一節(jié)點(diǎn)出現(xiàn)問題時(shí)候,即時(shí)改變數(shù)據(jù)流向,比如發(fā)送到數(shù)暫存服務(wù),避免Kafka中積累太多數(shù)據(jù),吞吐不平衡
通常流編排里面每個(gè)服務(wù)都有一個(gè)輸入及輸出,分別為input及sink,所以每個(gè)服務(wù)我們需要配置兩個(gè)topic,分別是input-topic output-topic,我們就在nacos里面添加輸入輸出配置
nacos配置項(xiàng)需要配置groupId,dataId,通常我們用服務(wù)名稱作為groupId,配置項(xiàng)的名稱作為dataId,如node1-server服務(wù)有一個(gè)input配置項(xiàng),配置如下:
完成其中一個(gè)服務(wù)的配置,其它服務(wù)參考下圖配置即可
@Configuration @NacosPropertySource(dataId = "input", groupId = "node1-server", autoRefreshed = true) // autoRefreshed=true指的是nacos中配置發(fā)生改變后會(huì)刷新,false代表只會(huì)使用服務(wù)啟動(dòng)時(shí)候讀取到的值 @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; } }
服務(wù)的輸入需要在服務(wù)啟動(dòng)時(shí)候創(chuàng)建消費(fèi)者,在topic發(fā)生改變時(shí)候重新創(chuàng)建消費(fèi)者,移除舊topic的消費(fèi)者,輸出是業(yè)務(wù)驅(qū)動(dòng)的,無需監(jiān)聽改變,在每次發(fā)送時(shí)候讀取到的都是最新配置的topic,因?yàn)樵谏厦娴呐渲妙愔衋utoRefreshed = true,這個(gè)只會(huì)刷新nacosConfig中的配置值,服務(wù)需要知道配置改變?nèi)ヲ?qū)動(dòng)消費(fèi)的創(chuàng)建業(yè)務(wù),需要?jiǎng)?chuàng)建nacos配置監(jiān)聽
/** * 監(jiān)聽Nacos配置改變,創(chuàng)建消費(fèi)者,更新消費(fèi) */ @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; // 用于存放當(dāng)前消費(fèi)者使用的topic private String topic; // 用于執(zhí)行消費(fèi)者線程 private ExecutorService executorService; /** * 監(jiān)聽input */ @NacosConfigListener(dataId = "node1-server", groupId = "input") public void inputListener(String input) { // 這個(gè)監(jiān)聽觸發(fā)的時(shí)候 實(shí)際NacosConfig中input的值已經(jīng)是最新的值了 我們只是需要這個(gè)監(jiān)聽觸發(fā)我們更新消費(fèi)者的業(yè)務(wù) String inputTopic = nacosConfig.getInput(); // 我使用nacosConfig中讀取的原因是因?yàn)楸O(jiān)聽到內(nèi)容是input=xxxx而不是xxxx,如果使用需要自己截取一下,nacosConfig中的內(nèi)容框架會(huì)處理好,大家看一下第一張圖的配置內(nèi)容就明白了 // 先檢查當(dāng)前局部變量topic是否有值,有值代表是更新消費(fèi)者,沒有值只需要?jiǎng)?chuàng)建即可 if(topic != null) { // 停止舊的消費(fèi)者線程 executorService.shutdownNow(); executorService == null; } // 根據(jù)為新的topic創(chuàng)建消費(fèi)者 topic = inputTopic; ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(topic + "-pool-%d").build(); executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(2), threadFactory); // 執(zhí)行消費(fèi)業(yè)務(wù) executorService.execute(() -> consumer(topic)); } /** * 創(chuàng)建消費(fèi)者 */ 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(); // 執(zhí)行數(shù)據(jù)處理業(yè)務(wù) 省略業(yè)務(wù)實(shí)現(xiàn) String handleMessage = handle(message); // 處理完成后發(fā)送到下一個(gè)節(jié)點(diǎn) kafkaTemplate.send(nacosConfig.getSink(), handleMessage); } } consumer.commitAsync(); } } catch (Exception e) { LOGGER.error(e.getMessage(), e); } finally { try { consumer.commitSync(); } finally { consumer.close(); } } } }
以上是“實(shí)現(xiàn)SpringBoot+Nacos+Kafka微服務(wù)流編排的示例分析”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關(guān)知識,歡迎關(guān)注億速云行業(yè)資訊頻道!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。