一、什么是消息總線

相信大多數(shù)讀者之前都使用過各種各樣的消息隊(duì)列，例如RabbitMQ、kafka等等，消息總線和他的概念差不多，在微服務(wù)系統(tǒng)的架構(gòu)中，我們通常會(huì)使用輕量級(jí)的消息代理來 構(gòu)建一個(gè)共用的消息主題讓系統(tǒng)中所有的微服務(wù)都連接上來，由于該主題中產(chǎn)生的消息會(huì)被所有實(shí)例監(jiān)聽和消費(fèi)，所以 我們稱他們?yōu)橄⒖偩€。在總線上的各個(gè)實(shí)例都可以方便的廣播一些需要讓其他連接到該主題上的實(shí)例都知道的消息，例如配置的變更或者其他一些管理操作等。

二、整合消息總線實(shí)現(xiàn)配置自動(dòng)刷新

在上一篇博客中spring cloud config 中我們實(shí)現(xiàn)了微服務(wù)架構(gòu)中的分布式配置中心，但是存在一個(gè)問題就是，當(dāng)我們?cè)趃it上修改了配置以后，需要我們手動(dòng)通知每一個(gè)服務(wù)實(shí)例，這樣的操作在實(shí)例較多的項(xiàng)目中是會(huì)死人的，這樣的問題sping cloud 家族肯定也是會(huì)考慮到并且給出解決方案的，下面我們就來搞一下。

2.1 面向客戶端基本架構(gòu)

當(dāng)我們系統(tǒng)按照上圖啟動(dòng)以后，圖中的 serviceA的三個(gè)實(shí)例會(huì)請(qǐng)求Config Server以獲取配置，Config Server根據(jù)應(yīng)用配置的規(guī)則從Git倉庫中獲取配置信息并返回。

此時(shí)，如果我們想要修改serviceA的配置。首先，去git服務(wù)器上修改對(duì)應(yīng)的參數(shù)值，但是這樣并不會(huì)觸發(fā)serviceA實(shí)例的屬性更新。此時(shí)我們向?qū)嵗?發(fā)送post請(qǐng)求，此時(shí)，實(shí)例3就會(huì)將刷新請(qǐng)求發(fā)送到消息總線中，該消息事件會(huì)被serviceA的實(shí)例1和實(shí)例2從總線中獲取到，并重新從config server中獲取他們的配置信息，從而實(shí)現(xiàn)配置信息的動(dòng)態(tài)更新。

2.2 面向服務(wù)端的架構(gòu)

在之前的架構(gòu)中，服務(wù)的配置更新需要通過具體服務(wù)中的某個(gè)實(shí)例發(fā)送請(qǐng)求，再觸發(fā)對(duì)整個(gè)服務(wù)集群的配置更新。雖然能 傷心啊功能，但是 這樣的結(jié)果是，我們指定的應(yīng)用實(shí)例會(huì)不同于集群中的其他應(yīng)用 實(shí)例，這樣會(huì)增加集群內(nèi)容的復(fù)雜度，不利于將來的運(yùn)維工作。

三、利用kafka實(shí)現(xiàn)消息總線

3.1 Spring Boot 整合kafka

可以參考這篇文章

如果 spring boot 版本采用 2.2.5，則kafka版本使用2.4.0.RELEASE。

3.2 實(shí)現(xiàn)動(dòng)態(tài) 刷新

我們利用上一篇博客中的config 的兩個(gè)工程來進(jìn)行改造。

3.2.1 服務(wù)端改造

增加依賴：

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bus-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-bus</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

增加配置：

spring.kafka.bootstrap-servers=211.159.167.180:9092
spring.kafka.consumer.group-id=test-consumer-group
spring.cloud.bus.enabled=true
management.endpoints.web.exposure.include= *

關(guān)于management.endpoints.web.exposure.include= * 的配置需要注意

注意：

• __如果是yum的話 ‘’ 需要加 ‘ ’ 單引號(hào)*
• include: ‘*’ http://localhost:8769/actuator/bus-refresh 刷新所有微服務(wù)
• include: ‘refresh’ http://localhost:8769/actuator/bus-refresh 不能訪問

3.2.2 客戶端改造

增加依賴：

<dependency>
? ? ? ? ? ? <groupId>org.springframework.cloud</groupId>
? ? ? ? ? ? <artifactId>spring-cloud-starter-bus-kafka</artifactId>
? ? ? ? </dependency>

? ? ? ? <dependency>
? ? ? ? ? ? <groupId>org.springframework.cloud</groupId>
? ? ? ? ? ? <artifactId>spring-cloud-bus</artifactId>
? ? ? ? </dependency>

增加配置：

management.endpoints.web.exposure.include= *
spring.kafka.bootstrap-servers=211.159.167.180:9092
spring.cloud.bus.enabled=true

這樣就ok 了，啟動(dòng)項(xiàng)目以后，當(dāng)配置修改以后，我們 給服務(wù)端發(fā)發(fā)送POST請(qǐng)求:http://localhost:7071/actuator/bus-refresh

就可以實(shí)現(xiàn)動(dòng)態(tài)刷新：

完整項(xiàng)目地址：https://github.com/zhenghaoxiao/spring-cloud-in-action/tree/bus bus 分支

3.3 指定刷新范圍

在上面的例子中，我們通過向服務(wù)端請(qǐng)求/actuator/bus-refresh接口，從而觸發(fā)總線上所有服務(wù)實(shí)例刷新，但是在一些特殊場(chǎng)景下，我們希望可以刷新服務(wù)中某個(gè)具體實(shí)例的配置，Spring Cloud Bus 對(duì)這種場(chǎng)景也有很好的支持，/actuator/bus-refreshdestination=customers:9000 提供了一個(gè)destination參數(shù)，用來定位具體要刷新的應(yīng)用程序。當(dāng)我們調(diào)用帶有destination參數(shù)的 接口時(shí)，此時(shí)總線上的個(gè)應(yīng)用實(shí)例會(huì)根據(jù)destination屬性的值來判斷是否為自己的實(shí)例名，若符合才進(jìn)行配置刷新，若不符合就忽略該 消息。

到此這篇關(guān)于基于kafka實(shí)現(xiàn)Spring Cloud Bus消息總線的文章就介紹到這了,更多相關(guān)kafka消息總線Spring Cloud Bus內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論