一、什么是消息總線

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

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

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

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

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

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

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

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

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

3.1 Spring Boot 整合kafka

可以參考這篇文章

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

3.2 實現(xiàn)動態(tài) 刷新

我們利用上一篇博客中的config 的兩個工程來進(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的話 ‘’ 需要加 ‘ ’ 單引號*
• 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)配置修改以后，我們 給服務(wù)端發(fā)發(fā)送POST請求:http://localhost:7071/actuator/bus-refresh

就可以實現(xiàn)動態(tài)刷新：

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

3.3 指定刷新范圍

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

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

最新評論