Spring Cloud（十一）高可用的分布式配置中心 Spring Cloud Bus 消息總線集成（RabbitMQ）

2022-06-10 17:05 更新

上一篇文章，留了一個(gè)懸念,Config Client 實(shí)現(xiàn)配置的實(shí)時(shí)更新，我們可以使用 /refresh 接口觸發(fā)，如果所有客戶端的配置的更改，都需要手動(dòng)觸發(fā)客戶端 /refresh ，當(dāng)服務(wù)越來越多的時(shí)候，那豈不是維護(hù)成本很高，顯然不太合適，而使用Spring Cloud Bus 消息總線實(shí)現(xiàn)方案，可以優(yōu)雅的解決以上問題，那就是通過消息代理中間件RabbitMQ 加 Git的Webhooks來觸發(fā)配置的更新，那具體是怎么實(shí)現(xiàn)的，我會(huì)通過圖文形式介紹。

Spring Cloud Bus

Spring Cloud Bus 將分布式系統(tǒng)的節(jié)點(diǎn)通過輕量級消息代理連接起來。用于在集群中傳播狀態(tài)更改（例如配置更改事件）或其他管理指令。Spring Cloud Bus 的一個(gè)核心思想是通過分布式的啟動(dòng)器對 Spring Boot 應(yīng)用進(jìn)行擴(kuò)展，也可以用來建立一個(gè)或多個(gè)應(yīng)用之間的通信頻道。目前唯一實(shí)現(xiàn)的方式是用 AMQP 消息代理作為通道，但是相同的基本功能集（還有一些取決于傳輸）在其他傳輸?shù)穆肪€圖上

消息總線

消息總線是一種通信工具，可以在機(jī)器之間互相傳輸消息、文件等。消息總線扮演著一種消息路由的角色，擁有一套完備的路由機(jī)制來決定消息傳輸方向。發(fā)送段只需要向消息總線發(fā)出消息而不用管消息被如何轉(zhuǎn)發(fā)。 Spring cloud bus 通過輕量消息代理連接各個(gè)分布的節(jié)點(diǎn)。管理和傳播所有分布式項(xiàng)目中的消息，本質(zhì)是利用了MQ的廣播機(jī)制在分布式的系統(tǒng)中傳播消息，目前常用的有Kafka和RabbitMQ 。 下面是一個(gè)配置中心刷新配置的例子

架構(gòu)圖

[圖片來源于網(wǎng)絡(luò)如有侵權(quán)請私信刪除]

  • 1、提交代碼觸發(fā)post請求給bus/refresh
  • 2、server端接收到請求并發(fā)送給Spring Cloud Bus
  • 3、Spring Cloud bus接到消息并通知給其它客戶端
  • 4、其它客戶端接收到通知，請求Server端獲取最新配置
  • 5、全部客戶端均獲取到最新的配置

消息代理

消息代理（Message Broker）是一種消息驗(yàn)證、傳輸、路由的架構(gòu)模式。消息代理是一個(gè)中間件產(chǎn)品，它的核心是一個(gè)消息的路由程序，用來實(shí)現(xiàn)接收和分發(fā)消息，并根據(jù)設(shè)定好的消息處理流來轉(zhuǎn)發(fā)給正確的應(yīng)用。它包括獨(dú)立的通信和消息傳遞協(xié)議，能夠?qū)崿F(xiàn)組織內(nèi)部和組織間的網(wǎng)絡(luò)通信。設(shè)計(jì)代理的目的就是為了能夠從應(yīng)用程序中傳入消息，并執(zhí)行一些特別的操作。

和組織間的網(wǎng)絡(luò)通信。設(shè)計(jì)代理的目的就是為了能夠從應(yīng)用程序中傳入消息，并執(zhí)行一些特別的操作。 現(xiàn)有的消息代理開源產(chǎn)品：

  • ActiveMQ
  • Kafka
  • RabbitMQ
  • RocketMQ

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

RabbitMQ簡介

RabbitMQ是一個(gè)開源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。

AMQP，即Advanced message Queuing Protocol，高級消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無需知道消息使用者的存在，反之亦然。

AMQP的主要特征是面向消息、隊(duì)列、路由（包括點(diǎn)對點(diǎn)和發(fā)布/訂閱）、可靠性、安全。

Github：https://github.com/rabbitmq
官網(wǎng)地址：http://www.rabbitmq.com

安裝RabbitMQ

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

CentOs7.3 搭建 RabbitMQ 3.6 單機(jī)服務(wù):
https://segmentfault.com/a/1190000010693696

CentOs7.3 搭建 RabbitMQ 3.6 Cluster 集群服務(wù):
https://segmentfault.com/a/1190000010702020

Spring Boot 中使用 RabbitMQ: https://segmentfault.com/a/1190000011577243

準(zhǔn)備工作

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

把上一篇，示例代碼下載，才可以進(jìn)行一下的操作，下載地址在文章末尾

  • 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

在項(xiàng)目spring-cloud-config-server 進(jìn)行以下操作

添加依賴

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

添加配置

在 application.properties 添加以下配置.關(guān)閉安全認(rèn)證

RabbitMQ 的 ymq用戶是手動(dòng)創(chuàng)建的，具體閱讀上面 安裝RabbitMQ 部分



#關(guān)閉刷新安全認(rèn)證
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

修改第上一篇文章項(xiàng)目

  • 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 添加以下配置.關(guān)閉安全認(rèn)證

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

測試服務(wù)

啟動(dòng)RabbitMQ

啟動(dòng)MQ服務(wù)

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

查看MQ狀態(tài)

$ 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/rabbit@node6.log
Dec 29 17:44:08 node6 rabbitmq-server[2948]: ######  ##        /var/log/rabbitmq/rabbit@node6-sasl.log
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]#

啟動(dòng)項(xiàng)目

按照順序依次啟動(dòng)項(xiàng)目

  • 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

啟動(dòng)該工程后，訪問服務(wù)注冊中心，查看服務(wù)是否都已注冊成功：http://127.0.0.1:8761/

查看服務(wù)注冊情況

Exchanges

檢查 springCloudBus

任何發(fā)送到Fanout Exchange 的消息都會(huì)被轉(zhuǎn)發(fā)到與該Exchange綁定(Binding)的所有springCloudBus 隊(duì)列Queue上。

檢查Queues

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

檢查 springCloudBus

修改配置

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

修改git 倉庫配置

查看 Config Server

通過 Postman 發(fā)送 GET 請求到：http://localhost:8888/springCloudConfig/dev/master 查看 Config Server 是否是最新的值

Config Server 已經(jīng)是新的值

查看 Config Client

命令窗口，通過curl http://127.0.0.1:9000/hello 訪問服務(wù)，或者在瀏覽器訪問http://127.0.0.1:9000/hello F5 刷新

發(fā)現(xiàn)沒有得到最新的值

因?yàn)槲覀儧]有主動(dòng)觸發(fā)Config Server bus/refresh接口

訪問服務(wù)

刷新配置

通過 Postman 發(fā)送 POST請求到：http://localhost:8888/bus/refresh ，我們可以看到以下內(nèi)容：

注意是 POST 請求

訪問：http://localhost:8888/bus/refresh

三個(gè)Config Client 客戶端控制臺(tái)，分別會(huì)打印以下內(nèi)容意思就是，收到遠(yuǎn)程更新請求，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://ymq@192.168.252.126:5672/, localPort= 64885]

再次查看 Config Client

訪問：http://localhost:8081/ ,http://localhost:8082/ ,http://localhost:8083/ 已經(jīng)刷新了配置

訪問服務(wù)

擴(kuò)展閱讀

Git webhooks

現(xiàn)在雖然可以不用重啟服務(wù)就更新配置了，但還是需要我們手動(dòng)操作，這樣還是不可取的。所以，這里就要用到Git的webhooks來達(dá)到自動(dòng)更新配置。

打開git上配置倉庫的地址，添加webhooks，上面Payload URL我寫的域名，當(dāng)然我沒有部署，上面的Payload URL就填寫我們的配置中心觸發(fā)刷新的地址，當(dāng)然這里不能寫localhost啦，要外網(wǎng)訪問地址才行。

還有這里面有個(gè)Secret的秘鑰驗(yàn)證，如果這里填寫的話，在配置文件上要寫上encrypt.key與之對應(yīng)。

添加webhooks

局部刷新

某些場景下（例如灰度發(fā)布），我們可能只想刷新部分微服務(wù)的配置，此時(shí)可通過/bus/refresh端點(diǎn)的destination參數(shù)來定位要刷新的應(yīng)用程序。

例如：/bus/refresh?destination=customers:8000，這樣消息總線上的微服務(wù)實(shí)例就會(huì)根據(jù)destination參數(shù)的值來判斷是否需要要刷新。其中，customers:8000指的是各個(gè)微服務(wù)的ApplicationContext ID。

destination參數(shù)也可以用來定位特定的微服務(wù)。例如：/bus/refresh?destination=customers:**，這樣就可以觸發(fā)customers微服務(wù)所有實(shí)例的配置刷新。

跟蹤總線事件

一些場景下，我們可能希望知道Spring Cloud Bus事件傳播的細(xì)節(jié)。此時(shí)，我們可以跟蹤總線事件（RemoteApplicationEvent的子類都是總線事件）。

跟蹤總線事件非常簡單，只需設(shè)置spring.cloud.bus.trace.enabled=true，這樣在/bus/refresh端點(diǎn)被請求后，訪問/trace端點(diǎn)就可獲得類似如下的結(jié)果：

發(fā)送 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
  • Email：admin@souyunku.com
  • 版權(quán)歸作者所有，轉(zhuǎn)載請注明出處

  • Wechat：關(guān)注公眾號，搜云庫技術(shù)團(tuán)隊(duì)，專注于開發(fā)技術(shù)的研究與知識(shí)分享

    關(guān)注公眾號-搜云庫

以上內(nèi)容是否對您有幫助：
在線筆記
App下載
App下載

掃描二維碼

下載編程獅App

公眾號
微信公眾號

編程獅公眾號