好了現在我們接著上一篇的隨筆，繼續來講。上一篇我們講到，我們如果要去更新所有微服務的配置，在不重啟的情況下去更新配置，只能依靠spring cloud config了，但是，是我們要一個服務一個服務的發送post請求，
我們能受的了嗎？這比之前的沒配置中心好多了，那么我們如何繼續避免挨個挨個的向服務發送post請求來告知服務，你的配置信息改變了，需要及時修改內存中的配置信息。
這時候我們就不要忘記消息隊列的發布訂閱模型。讓所有為服務來訂閱這個事件，當這個事件發生改變了，就可以通知所有微服務去更新它們的內存中的配置信息。這時bus消息總線就能解決，你只需要在springcloud config server端發出refresh，就可以觸發所有微服務更新了。
如下架構圖所示：
spring cloud bus除了支持rabbitmq的自動化配置之外，還支持現在被廣泛應用的kafka。在本文中，我們將搭建一個kafka的本地環境，并通過它來嘗試使用spring cloud bus對kafka的支持，實現消息總線的功能。
kafka使用scala實現，被用作linkedin的活動流和運營數據處理的管道，現在也被諸多互聯網企業廣泛地用作為數據流管道和消息系統。
kafak架構圖如下：
kafka是基于消息發布/訂閱模式實現的消息系統，其主要設計目標如下：
1.消息持久化：以時間復雜度為o(1)的方式提供消息持久化能力，即使對tb級以上數據也能保證常數時間復雜度的訪問性能。
2.高吞吐：在廉價的商用機器上也能支持單機每秒100k條以上的吞吐量
3.分布式：支持消息分區以及分布式消費，并保證分區內的消息順序
4.跨平臺：支持不同技術平臺的客戶端（如：java、php、python等）
5.實時性：支持實時數據處理和離線數據處理
6.伸縮性：支持水平擴展
kafka中涉及的一些基本概念：
1.broker：kafka集群包含一個或多個服務器，這些服務器被稱為broker。
2.topic：邏輯上同rabbit的queue隊列相似，每條發布到kafka集群的消息都必須有一個topic。（物理上不同topic的消息分開存儲，邏輯上一個topic的消息雖然保存于一個或多個broker上，但用戶只需指定消息的topic即可生產或消費數據而不必關心數據存于何處）
3.partition：partition是物理概念上的分區，為了提供系統吞吐率，在物理上每個topic會分成一個或多個partition，每個partition對應一個文件夾（存儲對應分區的消息內容和索引文件）。
4.producer：消息生產者，負責生產消息并發送到kafka broker。
5.consumer：消息消費者，向kafka broker讀取消息并處理的客戶端。
6.consumer group：每個consumer屬于一個特定的組（可為每個consumer指定屬于一個組，若不指定則屬于默認組），組可以用來實現一條消息被組內多個成員消費等功能。
可以從kafka的架構圖看到kafka是需要zookeeper支持的，你需要在你的kafka配置里面指定zookeeper在哪里，它是通過zookeeper做一些可靠性的保證，做broker的主從，我們還要知道kafka的消息是以topic形式作為組織的，producers發送topic形式的消息，consumer是按照組來分的，所以，一組consumers都會都要同樣的topic形式的消息。在服務端，它還做了一些分片，那么一個topic可能分布在不同的分片上面，方便我們拓展部署多個機器，kafka是天生分布式的。這里為了演示，我們只需要用它的默認配置，在windows上做個小demo即可。
我們這里主要針對spring cloud bus對kafka的支持，實現消息總線的功能，具體的kafka，rabbitmq消息隊列希望自己去找資料來學習一下。有了一些概念的支持后，我們進行一些demo。
如下：首先新建一個springcloud-config-client1模塊，方便我們進行測試所引入的依賴如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-actuator</artifactid> </dependency> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-web</artifactid> </dependency> <dependency> <groupid>org.springframework.cloud</groupid> <artifactid>spring-cloud-starter-config</artifactid> <version>1.4.0.release</version> </dependency> <dependency> <groupid>org.springframework.cloud</groupid> <artifactid>spring-cloud-starter-eureka</artifactid> <version>1.3.5.release</version> </dependency> <dependency> <groupid>org.springframework.cloud</groupid> <artifactid>spring-cloud-starter-bus-kafka</artifactid> <version>1.3.2.release</version> </dependency> |
接著要注意一下，client1的配置文件要改為bootstrap.yml，因為這種配置格式，是優先加載的，上一篇隨筆有講過，client1的配置如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
server: port: 7006spring: application: name: cloud-config cloud: config:#啟動什么環境下的配置，dev 表示開發環境，這跟你倉庫的文件的后綴有關，比如，倉庫配置文件命名格式是cloud-config-dev.properties,所以profile 就要寫dev profile: dev discovery: enabled: true#這個名字是config server端的服務名字，不能瞎寫。 service-id: config-server#注冊中心eureka: client: service-url: defaultzone: http://localhost:8888/eureka/,http://localhost:8889/eureka/#是否需要權限拉去，默認是true,如果不false就不允許你去拉取配置中心server更新的內容management: security: enabled: false |
接著啟動類如下：
|
1
2
3
4
5
6
7
8
|
@springbootapplication@enablediscoveryclientpublic class client1application { public static void main(string[] args) { springapplication.run(client1application.class, args); }} |
接著將client中的testcontroller賦值一份到client1中，代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
@restcontroller//這里面的屬性有可能會更新的，git中的配置中心變化的話就要刷新，沒有這個注解內，配置就不能及時更新@refreshscopepublic class testcontroller { @value("${name}") private string name; @value("${age}") private integer age; @requestmapping("/test") public string test(){ return this.name+this.age; }} |
接著還要在先前的隨筆中的模塊中的config server加入如下配置：
|
1
2
3
4
|
#是否需要權限拉去，默認是true,如果不false就不允許你去拉取配置中心server更新的內容management: security: enabled: false |
接著還要做一點就是，在config-client,config-client1,和config-server都要引入kafka的依賴，如下：
|
1
2
3
4
5
|
<dependency> <groupid>org.springframework.cloud</groupid> <artifactid>spring-cloud-starter-bus-kafka</artifactid> <version>1.3.2.release</version> </dependency> |
我們工程準備好了，暫時先放在這里，下面進行kafka的安裝下載，首先我們去kafka官網kafka.apache.org/downloads 下來官網推薦的版本，
首先我們進到下載好的kafka目錄中kafka_2.11-1.1.0\bin\windows 下編輯kafka-run-class.bat如下：
找到這條配置 如下：
可以看到%classpath%沒有雙引號，
因此用雙引號括起來，不然啟動不起來的，報你jdk沒安裝好，修改后如下：
接著，打開config文件夾中的server.properties配置如下：
可以看到是連接到本地的zookeeper就行了。
接著我們進行先啟動zookeeper，再啟動kafka，如下：
當看到上面的信息證明啟動zookeeper啟動成功。、
接下來再開一個cmd啟動kafka，如下：
看到這些信息說明kafka啟動成功了
好了，接下來把前面的工程，兩個注冊中心，一個springcloud-config-server,兩個springcloud-config-client,springcloud-config-client1啟動起來，
可以看到springcloudbus是在0分片上，如果兩個config-client啟動都出現上面信息，證明啟動成功了。
好了現在我們進行訪問一下config-server端，如下：
再訪問兩個client,如下：
好了，好戲開始了，現在我們去git倉庫上修改配置中心的文件，將年齡改為24，如下：
接下來，我們我們用refresh刷新配置服務端配置，通知兩個client去更新內存中的配置信息。用postman發送localhost:7000/bus/refresh,如下：
可以看到沒有返回什么信息，但是不要擔心，這是成功的通知所有client去更新了內存中的信息了。
接著我們分別重新請求config-server,兩個client,刷新頁面，結果如下：
兩個client如下：
可以看到所有client自動更新內存中的配置信息了。
到目前為止，上面都是刷新說有的配置的信息的，如果我們想刷新某個特定服務的配置信息也是可以的。我們可以指定刷新范圍，如下：
指定刷新范圍
上面的例子中，我們通過向服務實例請求spring cloud bus的/bus/refresh接口，從而觸發總線上其他服務實例的/refresh。但是有些特殊場景下（比如：灰度發布），我們希望可以刷新微服務中某個具體實例的配置。
spring cloud bus對這種場景也有很好的支持：/bus/refresh接口還提供了destination參數，用來定位具體要刷新的應用程序。比如，我們可以請求/bus/refresh?destination=服務名字:9000，此時總線上的各應用實例會根據destination屬性的值來判斷是否為自己的實例名，
若符合才進行配置刷新，若不符合就忽略該消息。
destination參數除了可以定位具體的實例之外，還可以用來定位具體的服務。定位服務的原理是通過使用spring的pathmatecher（路徑匹配）來實現，比如：/bus/refresh?destination=customers:**，該請求會觸發customers服務的所有實例進行刷新。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：http://www.cnblogs.com/huangjuncong/p/9077099.html
