基于kafka怎么實現Spring?Cloud?Bus消息總線

這篇文章主要介紹“基于kafka怎么實現Spring Cloud Bus消息總線”的相關知識，小編通過實際案例向大家展示操作過程，操作方法簡單快捷，實用性強，希望這篇“基于kafka怎么實現Spring Cloud Bus消息總線”文章能幫助大家解決問題。

一、什么是消息總線

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

二、整合消息總線實現配置自動刷新

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

2.1 面向客戶端基本架構

當我們系統按照上圖啟動以后，圖中的 serviceA的三個實例會請求Config Server以獲取配置，Config Server根據應用配置的規則從Git倉庫中獲取配置信息并返回。

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

2.2 面向服務端的架構

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

三、利用kafka實現消息總線

3.1 Spring Boot 整合kafka

可以參考這篇文章

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

3.2 實現動態 刷新

我們利用上一篇博客中的config 的兩個工程來進行改造。

3.2.1 服務端改造

增加依賴：

<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= *

關于management.endpoints.web.exposure.include= * 的配置需要注意

注意：

• __如果是yum的話 ‘’ 需要加 ‘ ’ 單引號*

• include: ‘*’ http://localhost:8769/actuator/bus-refresh 刷新所有微服務

• 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 了，啟動項目以后，當配置修改以后，我們 給服務端發發送POST請求:http://localhost:7071/actuator/bus-refresh

就可以實現動態刷新。

3.3 指定刷新范圍

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

關于“基于kafka怎么實現Spring Cloud Bus消息總線”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識，可以關注億速云行業資訊頻道，小編每天都會為大家更新不同的知識點。