Spring Cloud（十一）高可用的分散式配置中心 Spring Cloud Bus 訊息匯流排整合（RabbitMQ）

阿新 • • 發佈：2022-04-28

上一篇文章，留了一個懸念,Config Client 實現配置的實時更新，我們可以使用 /refresh 介面觸發，如果所有客戶端的配置的更改，都需要手動觸發客戶端 /refresh ，當服務越來越多的時候，那豈不是維護成本很高，顯然不太合適，而使用Spring Cloud Bus 訊息匯流排實現方案，可以優雅的解決以上問題，那就是通過訊息代理中介軟體RabbitMQ 加 Git的Webhooks來觸發配置的更新，那具體是怎麼實現的，我會通過圖文形式介紹。

Spring Cloud Bus

Spring Cloud Bus 將分散式系統的節點通過輕量級訊息代理連線起來。用於在叢集中傳播狀態更改（例如配置更改事件）或其他管理指令。Spring Cloud Bus

的一個核心思想是通過分散式的啟動器對 Spring Boot 應用進行擴充套件，也可以用來建立一個或多個應用之間的通訊頻道。目前唯一實現的方式是用 AMQP 訊息代理作為通道，但是相同的基本功能集（還有一些取決於傳輸）在其他傳輸的路線圖上

訊息匯流排

訊息匯流排是一種通訊工具，可以在機器之間互相傳輸訊息、檔案等。訊息匯流排扮演著一種訊息路由的角色，擁有一套完備的路由機制來決定訊息傳輸方向。傳送段只需要向訊息匯流排發出訊息而不用管訊息被如何轉發。 Spring cloud bus 通過輕量訊息代理連線各個分佈的節點。管理和傳播所有分散式專案中的訊息，本質是利用了MQ的廣播機制在分散式的系統中傳播訊息，目前常用的有Kafka

和RabbitMQ 。下面是一個配置中心重新整理配置的例子

[圖片來源於網路如有侵權請私信刪除]

1、提交程式碼觸發post請求給bus/refresh
2、server端接收到請求併發送給Spring Cloud Bus
3、Spring Cloud bus接到訊息並通知給其它客戶端
4、其它客戶端接收到通知，請求Server端獲取最新配置
5、全部客戶端均獲取到最新的配置

訊息代理

訊息代理（Message Broker）是一種訊息驗證、傳輸、路由的架構模式。訊息代理是一箇中間件產品，它的核心是一個訊息的路由程式，用來實現接收和分發訊息，並根據設定好的訊息處理流來轉發給正確的應用。它包括獨立的通訊和訊息傳遞協議，能夠實現組織內部和組織間的網路通訊。設計代理的目的就是為了能夠從應用程式中傳入訊息，並執行一些特別的操作。

和組織間的網路通訊。設計代理的目的就是為了能夠從應用程式中傳入訊息，並執行一些特別的操作。現有的訊息代理開源產品：

ActiveMQ
Kafka
RabbitMQ
RocketMQ

目前Spring Cloud Bus 支援 RabbitMQ 和 Kafka，spring-cloud-starter-bus-amqp 、spring-cloud-starter-bus-kafka

RabbitMQ簡介

RabbitMQ是一個開源的AMQP實現，伺服器端用Erlang語言編寫，支援多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支援AJAX。用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。

AMQP，即Advanced message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道訊息使用者的存在，反之亦然。

AMQP的主要特徵是面向訊息、佇列、路由（包括點對點和釋出/訂閱）、可靠性、安全。

Github：https://github.com/rabbitmq 官網地址：http://www.rabbitmq.com

安裝RabbitMQ

安裝RabbitMQ 可以參考之前的文章

CentOs7.3 搭建 RabbitMQ 3.6 單機服務: https://cloud.tencent.com/developer/article/1041042

CentOs7.3 搭建 RabbitMQ 3.6 Cluster 叢集服務: https://cloud.tencent.com/developer/article/1041143

Spring Boot 中使用 RabbitMQ: https://cloud.tencent.com/developer/article/1041365

準備工作

以下專案修改不做過多解釋，部分程式碼不再展示，請閱讀上篇文章，Spring Cloud（十）高可用的分散式配置中心 Spring Cloud Config 中使用 Refresh：http://www.ymq.io/2017/12/23/spring-cloud-config-eureka-refresh/

把上一篇，示例程式碼下載，才可以進行一下的操作，下載地址在文章末尾

spring-cloud-eureka-service spring-cloud-config-server spring-cloud-eureka-provider-1 spring-cloud-eureka-provider-2 spring-cloud-eureka-provider-3 spring-cloud-feign-consumer

Config Server

在專案spring-cloud-config-server 進行以下操作

新增依賴

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

新增配置

在 application.properties 新增以下配置.關閉安全認證

RabbitMQ 的 ymq使用者是手動建立的，具體閱讀上面安裝RabbitMQ 部分

#關閉重新整理安全認證
management.security.enabled=false

spring.rabbitmq.host=192.168.252.126
spring.rabbitmq.port=5672
spring.rabbitmq.username=ymq
spring.rabbitmq.password=123456

Config Client

修改第上一篇文章專案

spring-cloud-eureka-provider-1 spring-cloud-eureka-provider-2 spring-cloud-eureka-provider-3

新增依賴

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

新增配置

在 application.properties 新增以下配置.關閉安全認證

spring.rabbitmq.host=192.168.252.126
spring.rabbitmq.port=5672
spring.rabbitmq.username=ymq
spring.rabbitmq.password=123456

測試服務

啟動RabbitMQ

啟動MQ服務

$ service rabbitmq-server start
Redirecting to /bin/systemctl start  rabbitmq-server.service

檢視MQ狀態

$ service rabbitmq-server status

[root@node6 rabbitmq]# service rabbitmq-server status
Redirecting to /bin/systemctl status  rabbitmq-server.service
● rabbitmq-server.service - RabbitMQ broker
   Loaded: loaded (/usr/lib/systemd/system/rabbitmq-server.service; disabled; vendor preset: disabled)
   Active: active (running) since Fri 2017-12-29 17:44:10 CST; 9min ago
  Process: 2814 ExecStop=/usr/sbin/rabbitmqctl stop (code=exited, status=0/SUCCESS)
 Main PID: 2948 (beam)
   Status: "Initialized"
   CGroup: /system.slice/rabbitmq-server.service
           ├─2948 /usr/lib64/erlang/erts-8.0.3/bin/beam -W w -A 64 -P 1048576 -t 5000000 -stbt db -zdbbl 32000 -K true -- -root /usr/lib64/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa /usr...
           ├─3131 /usr/lib64/erlang/erts-8.0.3/bin/epmd -daemon
           ├─3233 erl_child_setup 1024
           ├─3240 inet_gethost 4
           └─3241 inet_gethost 4

Dec 29 17:44:08 node6 rabbitmq-server[2948]: RabbitMQ 3.6.10. Copyright (C) 2007-2017 Pivotal Software, Inc.
Dec 29 17:44:08 node6 rabbitmq-server[2948]: ##  ##      Licensed under the MPL.  See http://www.rabbitmq.com/
Dec 29 17:44:08 node6 rabbitmq-server[2948]: ##  ##
Dec 29 17:44:08 node6 rabbitmq-server[2948]: ##########  Logs: /var/log/rabbitmq/[email protected]
Dec 29 17:44:08 node6 rabbitmq-server[2948]: ######  ##        /var/log/rabbitmq/[email protected]
Dec 29 17:44:08 node6 rabbitmq-server[2948]: ##########
Dec 29 17:44:08 node6 rabbitmq-server[2948]: Starting broker...
Dec 29 17:44:10 node6 rabbitmq-server[2948]: systemd unit for activation check: "rabbitmq-server.service"
Dec 29 17:44:10 node6 systemd[1]: Started RabbitMQ broker.
Dec 29 17:44:10 node6 rabbitmq-server[2948]: completed with 6 plugins.
[root@node6 rabbitmq]#

啟動專案

按照順序依次啟動專案

spring-cloud-eureka-service spring-cloud-config-server spring-cloud-eureka-provider-1 spring-cloud-eureka-provider-2 spring-cloud-eureka-provider-3 spring-cloud-feign-consumer

啟動該工程後，訪問服務註冊中心，檢視服務是否都已註冊成功：http://127.0.0.1:8761/

Exchanges

任何傳送到Fanout Exchange 的訊息都會被轉發到與該Exchange繫結(Binding)的所有springCloudBus 佇列Queue上。

檢查Queues

瀏覽器開啟：http://192.168.252.128:15672/

修改配置

修改Git倉庫配置，在 content=hello dev 後面加上 Spring Cloud Bus Test

檢視 Config Server

通過 Postman 傳送 GET 請求到：http://localhost:8888/springCloudConfig/dev/master 檢視 Config Server 是否是最新的值

檢視 Config Client

命令視窗，通過curl http://127.0.0.1:9000/hello 訪問服務，或者在瀏覽器訪問http://127.0.0.1:9000/hello F5 重新整理

發現沒有得到最新的值

因為我們沒有主動觸發Config Server bus/refresh介面

重新整理配置

通過 Postman 傳送 POST請求到：http://localhost:8888/bus/refresh ，我們可以看到以下內容：

注意是 PSOT 請求

三個Config Client 客戶端控制檯，分別會列印以下內容意思就是，收到遠端更新請求，config.client，KEYS 重新整理, key 是 content

2017-12-29 18:38:49.023  INFO 28944 --- [jeTgrKRGzgj9g-1] o.s.cloud.bus.event.RefreshListener      : Received remote refresh request. Keys refreshed [config.client.version, content]
2017-12-29 18:38:49.025  INFO 28944 --- [nfoReplicator-0] com.netflix.discovery.DiscoveryClient    : DiscoveryClient_EUREKA-PROVIDER/localhost:eureka-provider:8081: registering service...
2017-12-29 18:38:49.035  INFO 28944 --- [nfoReplicator-0] com.netflix.discovery.DiscoveryClient    : DiscoveryClient_EUREKA-PROVIDER/localhost:eureka-provider:8081 - registration status: 204
2017-12-29 18:38:49.067  INFO 28944 --- [jeTgrKRGzgj9g-1] o.s.a.r.c.CachingConnectionFactory       : Created new connection: SpringAMQP#31e87320:0/SimpleConnection@39151d4e [delegate=amqp://[email protected]:5672/, localPort= 64885]

再次檢視 Config Client

訪問：http://localhost:8081/ ,http://localhost:8082/ ,http://localhost:8083/ 已經重新整理了配置

擴充套件閱讀

Git webhooks

現在雖然可以不用重啟服務就更新配置了，但還是需要我們手動操作，這樣還是不可取的。所以，這裡就要用到Git的webhooks來達到自動更新配置。

開啟git上配置倉庫的地址，新增webhooks，上面Payload URL我寫的域名，當然我沒有部署，上面的Payload URL就填寫我們的配置中心觸發重新整理的地址，當然這裡不能寫localhost啦，要外網訪問地址才行。

還有這裡面有個Secret的祕鑰驗證，如果這裡填寫的話，在配置檔案上要寫上encrypt.key與之對應。

區域性重新整理

某些場景下（例如灰度釋出），我們可能只想重新整理部分微服務的配置，此時可通過/bus/refresh端點的destination引數來定位要重新整理的應用程式。

例如：/bus/refresh?destination=customers:8000，這樣訊息總線上的微服務例項就會根據destination引數的值來判斷是否需要要重新整理。其中，customers:8000指的是各個微服務的ApplicationContext ID。

destination引數也可以用來定位特定的微服務。例如：/bus/refresh?destination=customers:**，這樣就可以觸發customers微服務所有例項的配置重新整理。

跟蹤匯流排事件

一些場景下，我們可能希望知道Spring Cloud Bus事件傳播的細節。此時，我們可以跟蹤匯流排事件（RemoteApplicationEvent的子類都是匯流排事件）。

跟蹤匯流排事件非常簡單，只需設定spring.cloud.bus.trace.enabled=true，這樣在/bus/refresh端點被請求後，訪問/trace端點就可獲得類似如下的結果：

傳送 GET請求到：http://localhost:8888/trace

[
    {
        "timestamp": 1514543931362,
        "info": {
            "method": "GET",
            "path": "/eureka-provider/dev/master",
            "headers": {
                "request": {
                    "accept": "application/json, application/*+json",
                    "user-agent": "Java/1.8.0_112",
                    "host": "localhost:8888",
                    "connection": "keep-alive"
                },
                "response": {
                    "X-Application-Context": "config-server:8888",
                    "Content-Type": "application/json;charset=UTF-8",
                    "Transfer-Encoding": "chunked",
                    "Date": "Fri, 29 Dec 2017 10:38:51 GMT",
                    "status": "200"
                }
            },
            "timeTaken": "6002"
        }
    },
    {
        "timestamp": 1514543927451,
        "info": {
            "method": "GET",
            "path": "/eureka-provider/dev/master",
            "headers": {
                "request": {
                    "accept": "application/json, application/*+json",
                    "user-agent": "Java/1.8.0_112",
                    "host": "localhost:8888",
                    "connection": "keep-alive"
                },
                "response": {
                    "X-Application-Context": "config-server:8888",
                    "Content-Type": "application/json;charset=UTF-8",
                    "Transfer-Encoding": "chunked",
                    "Date": "Fri, 29 Dec 2017 10:38:47 GMT",
                    "status": "200"
                }
            },
            "timeTaken": "4927"
        }
    },
    {
        "timestamp": 1514543925254,
        "info": {
            "method": "GET",
            "path": "/eureka-provider/dev/master",
            "headers": {
                "request": {
                    "accept": "application/json, application/*+json",
                    "user-agent": "Java/1.8.0_112",
                    "host": "localhost:8888",
                    "connection": "keep-alive"
                },
                "response": {
                    "X-Application-Context": "config-server:8888",
                    "Content-Type": "application/json;charset=UTF-8",
                    "Transfer-Encoding": "chunked",
                    "Date": "Fri, 29 Dec 2017 10:38:45 GMT",
                    "status": "200"
                }
            },
            "timeTaken": "2862"
        }
    },
    {
        "timestamp": 1514543923565,
        "info": {
            "method": "POST",
            "path": "/bus/refresh",
            "headers": {
                "request": {
                    "cache-control": "no-cache",
                    "postman-token": "0e497ec1-0c03-4dc2-bb61-ce2a266227d3",
                    "user-agent": "PostmanRuntime/7.1.1",
                    "accept": "*/*",
                    "host": "127.0.0.1:8888",
                    "accept-encoding": "gzip, deflate",
                    "content-length": "0",
                    "connection": "keep-alive"
                },
                "response": {
                    "X-Application-Context": "config-server:8888",
                    "status": "200"
                }
            },
            "timeTaken": "6616"
        }
    }
]

原始碼下載

GitHub：https://github.com/souyunku/spring-cloud-examples/tree/master/spring-cloud-config-bus-rabbitMQ

碼雲：https://gitee.com/souyunku/spring-cloud-examples/tree/master/spring-cloud-config-bus-rabbitMQ

Contact

作者：鵬磊
出處：http://www.souyunku.com/2017/12/24/spring-cloud-config-bus-rabbitMQ
Email：[email protected]
版權歸作者所有，轉載請註明出處
Wechat：關注公眾號，搜雲庫，專注於開發技術的研究與知識分享

Spring Cloud（十一）高可用的分散式配置中心 Spring Cloud Bus 訊息匯流排整合（RabbitMQ）

Spring Cloud Bus

訊息匯流排

訊息代理

RabbitMQ簡介

安裝RabbitMQ

準備工作

Config Server

新增依賴

新增配置

Config Client

新增依賴

新增配置

測試服務

啟動RabbitMQ

啟動專案

Exchanges

檢查Queues

修改配置

檢視 Config Server

檢視 Config Client

重新整理配置

再次檢視 Config Client

擴充套件閱讀

Git webhooks

區域性重新整理

跟蹤匯流排事件

原始碼下載

Contact

相關推薦