當前位置：首頁 > 前端技术 > javascript >内容正文

javascript

SpringCloud微服务架构,Config 分布式配置中心,Bus 消息总线, Stream 消息驱动,Sleuth+Zipkin 链路追踪

發布時間：2024/4/15 javascript 23 豆豆

生活随笔收集整理的這篇文章主要介紹了 SpringCloud微服务架构,Config 分布式配置中心,Bus 消息总线, Stream 消息驱动,Sleuth+Zipkin 链路追踪小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

Config分布式配置中心

Config 概述

概述
? Spring Cloud Config 解決了在分布式場景下多環境配置文件的管理和維護。
? 好處：
? 集中管理配置文件
? 不同環境不同配置，動態化的配置更新
? 配置信息改變時，不需要重啟即可更新配置信息到服務

Bus 消息總線

Bus 概述
? Spring Cloud Bus 是用輕量的消息中間件將分布式的節點連接起來，可以用于廣播配置文件的更改或者服務的監控管理。關鍵的思想就是，消息總線可以為微服務做監控，也可以實現應用程序之間相通信。
? Spring Cloud Bus 可選的消息中間件包括 RabbitMQ 和 Kafka 。

使用RabbitMQ做消息中間件

RabbitMQ
RabbitMQ 提供了 6 種工作模式：簡單模式、work queues、Publish/Subscribe 發布與訂閱模式、Routing
路由模式、Topics 主題模式、RPC 遠程調用模式（遠程調用，不太算 MQ；暫不作介紹）。

Bus 簡單項目案例(代碼中有詳細解釋)

步驟:

分別在 config-server 和 config-client中引入 bus依賴：bus-amqp

分別在 config-server 和 config-client中配置 RabbitMQ

在config-server中設置暴露監控斷點：bus-refresh

啟動測試

搭建一個Springcloud的父項目

父項目 pom.xml

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.fs</groupId><artifactId>study-springcloud</artifactId><version>1.0-SNAPSHOT</version><modules><module>fs-server-eureka-7001</module><module>fs-config-server</module><module>fs-config-provider-8001</module><module>fs-config-provider-8002</module><module>fs-stream-RabbitMQ-provider-8801</module><module>fs-stream-RabbitMQ-consumer-9901</module></modules><packaging>pom</packaging><dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>2.3.2.RELEASE</version><type>pom</type> <scope>import</scope></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>Hoxton.SR6</version><type>pom</type> <scope>import</scope></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-alibaba-dependencies</artifactId><version>2.2.1.RELEASE</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-server</artifactId><version>2.2.4.RELEASE</version></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId><version>2.2.4.RELEASE</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.20</version><scope>runtime</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>1.1.20</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.3.2</version></dependency> <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.12</version></dependency></dependencies></dependencyManagement><dependencies></dependencies></project>

fs-server-eureka-7001(請參考另一篇博文,服務治理,有詳細說明搭建)

fs-config-server 搭建configserver

pom.xml

application.yml

server:port: 3344 spring:application:name: fs-config-servercloud:# 配置git遠程廠庫地址config:server:git:uri: https://gitee.com/xiaofugitee/springcloud-config.git # 碼云倉庫地址# 搜索目錄 # search-paths: # - springcloud-config #倉庫名字 # username: xiaofuGitee # password: ***********# 跳過ssl驗證 # skip-ssl-validation: true# 讀取分支label: master# RabbitMQ相關配置rabbitmq:host: 192.168.93.132port: 5672username: rootpassword: rootvirtual-host: / #服務注冊到eureka eureka:client:service-url:defaultZone: http://localhost:7001/eureka #,http://fseureka7002.com:7002/eureka,http://fseureka7003.com:7003/eureka # 暴露bus刷新配置的端點,使用bus,只需要在config server中配置就可以 management:endpoints:web:exposure:include: 'bus-refresh' # 使用bus發送刷新的post請求 # 由于配置了bus消息,gitee上的配置文件進行更改,只需要發送一個3344post請求處處生效3355,3366的配置都會進行更新, # curl -X POST "http://localhost:3344/actuator/bus-refresh"

ConfigServer3344 主啟動

//開啟作為配置中心
@EnableConfigServer

package com.fs;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.config.server.EnableConfigServer; import org.springframework.cloud.netflix.eureka.EnableEurekaClient;@SpringBootApplication @EnableEurekaClient //開啟作為配置中心 @EnableConfigServer public class ConfigServer3344 {public static void main(String[] args) {SpringApplication.run(ConfigServer3344.class,args);} }

fs-config-provider-8001/8002 搭建服務提供,來動態獲取配置信息

8001與8002 代碼一模一樣,只是bootstrap的server.port端口一個是8001,一個是8002

pom.xml

bootstrap.yml

server:port: 8001 spring:application:name: fs-config-providercloud:config:label: master # 分支名稱name: config # 配置文件前綴名 - 前面的名稱profile: dev # 讀取后綴名就是讀取碼云倉庫master分支下的config-dev.yml配置文件 # uri: http://localhost:3344 # 配置中心地址# 從eureka中發現configserver的服務的uri 這里寫了,就不用寫uri了discovery:enabled: true # 是否從注冊中心獲取信息service-id: fs-config-server #　服務名稱# RabbitMQ相關配置rabbitmq:host: 192.168.93.132port: 5672username: rootpassword: rootvirtual-host: / eureka:client:service-url:defaultZone: http://localhost:7001/eureka #,http://fseureka7002.com:7002/eureka,http://fseureka7003.com:7003/eureka # 暴露bus刷新配置的端點才能動態刷新配置,使用bus,只需要在config server中配置就可以 #management: # endpoints: # web: # exposure: # include: 'bus-refresh' # include: '*' #暴露所有

application.yml空沒有任何代碼

ConfigProvider8001/8002 主啟動

package com.fs;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.netflix.eureka.EnableEurekaClient;@SpringBootApplication @EnableEurekaClient public class ConfigProvider8001 {public static void main(String[] args) {SpringApplication.run(ConfigProvider8001.class,args);} }

ConfigProviderController

@RefreshScope//配置刷新功能
//cmd手動刷新才能獲取到最新的配置信息 curl -X POST “http://localhost:8001/actuator/refresh”
//但是使用了BUS消息總線后,只需要在configserver中發送一次請求,所有綁定的微服務,都會被廣播到,
// 就不需要每個微服務都去發送請求,來獲取最新的配置信息

package com.fs.controller;import org.springframework.beans.factory.annotation.Value; import org.springframework.cloud.context.config.annotation.RefreshScope; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController;@RestController @RefreshScope//配置刷新功能 //cmd手動刷新才能獲取到最新的配置信息 curl -X POST "http://localhost:8001/actuator/refresh" //但是使用了BUS消息總線后,只需要在configserver中發送一次請求,所有綁定的微服務,都會被廣播到, // 就不需要每個微服務都去發送請求,來獲取最新的配置信息 public class ConfigProviderController {@Value("${server.port}")private String serverPort;//獲取云(git) 上的配置文件信息,來確保我們的的服務是否獲取到最新的@Value("${config.info}")private String configInfo;//輸出一下最新的配置信息@GetMapping("/configInfo")public String getConfigInfo(){return "serverPort:"+serverPort+"~~~configInfo:"+configInfo;} }

. 啟動測試

在giree 碼云上準備一個廠庫(廠庫名字與git地址記得在configserver的application.yml文件中修改)

首先啟動RabbitMQ

啟動服務

依次啟動:
1.fs-server-eureka-7001
2.fs-config-server
啟動configserver報錯,不管,實際上是連接到了的
org.eclipse.jgit.api.errors.TransportException: https://gitee.com/xiaofugitee/springcloud-config.git: Secure connection to https://gitee.com/xiaofugitee/springcloud-config.git could not be stablished because of SSL problems

3.fs-config-provider-8001
4,fs-config-provider-8002

打開瀏覽器測試

http://localhost:3344/config-dev.yml
http://localhost:8001/configInfo
http://localhost:8002/configInfo

我們去gitee上修改一下配置文件中的version

修改gitee上的配置文件信息后,再次打開瀏覽器測試

http://localhost:3344/config-dev.yml
http://localhost:8001/configInfo
http://localhost:8002/configInfo

由于配置了bus消息,gitee上的配置文件進行更改,只需要發送一個3344post請求處處生效3355,3366的配置都會進行更新,
** curl -X POST http://localhost:3344/actuator/bus-refresh **

發送post請求廣播通知后,再次瀏覽器測試:

http://localhost:3344/config-dev.yml
http://localhost:8001/configInfo
http://localhost:8002/configInfo

Stream 消息驅動

Stream 概述

? Spring Cloud Stream 是一個構建消息驅動微服務應用的框架。
? Stream 解決了開發人員無感知的使用消息中間件的問題，因為Stream對消息中間件的進一步封裝，可以做
到代碼層面對中間件的無感知，甚至于動態的切換中間件，使得微服務開發的高度解耦，服務可以關注更多
自己的業務流程。
? Spring Cloud Stream目前支持兩種消息中間件RabbitMQ和Kafka

Stream 組件

? Spring Cloud Stream 構建的應用程序與消息中間件之間是通過綁定器 Binder
相關聯的。綁定器對于應用程序而言起到了隔離作用，它使得不同消息中間件
的實現細節對應用程序來說是透明的。
? binding 是我們通過配置把應用和spring cloud stream 的 binder 綁定在一起
? output：發送消息 Channel，內置 Source接口
? input：接收消息 Channel，內置 Sink接口

小案列演示 Stream 消息驅動

fs-stream-RabbitMQ-provider-8801 Stream 消息生產者

Stream 消息生產者

創建消息生產者模塊，引入依賴 starter-stream-rabbit

編寫配置，定義 binder，和 bingings

定義消息發送業務類。添加 @EnableBinding(Source.class)，注入
MessageChannel output ，完成消息發送

編寫啟動類，測試

pom.xml

application.yml

server:port: 8801 spring:application:name: fs-stream-providercloud:stream:# 定義綁定,綁定到哪個消息中間件上binders: #在此處聲明配置要綁定的RabbitMQ的服務信息fs_Rabbit: # 表示定義的名稱,用binding整合,隨便自定義type: rabbit # 消息組件類型environment: #設置RabbitMQ的相關配置spring:rabbitmq:host: 192.168.93.132port: 5672username: rootpassword: rootvirtual-host: /bindings: #服務的整合處理output: # 這個名字是一個管道的名稱,使用默認的outputdefault-binder: fs_Rabbit # 設置要綁定的消息服務的具體設置, 若idea爆紅沒關系destination: fs_Exchange #表示要使用的Exchange交換機名稱自定義消息發送的目的地content-type: application/json # 設置消息類型,本次為json 文本設置為test/plain#eureka: # client: # service-url: # defaultZone: http://fseureka7001.com:7001/eureka,http://fseureka7002.com:7002/eureka,http://fseureka7003.com:7003/eureka # instance: # lease-renewal-interval-in-seconds: 2 # 設置心跳的時間間隔(默認30秒) # lease-expiration-duration-in-seconds: 5 # 如果現在超過了5秒間隔(默認是90秒) # instance-id: send-8801.com # 在消息列表顯示主機名稱 # prefer-ip-address: true # 訪問的路徑變ip地址

StreamMain8801 主啟動

package com.fs.springcloud;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication public class StreamMain8801 {public static void main(String[] args) {SpringApplication.run(StreamMain8801.class,args);} }

IMessageProvider 服務提供接口

package com.fs.springcloud.service;public interface IMessageProvider {public String send(); }

IMessageProviderImpl 服務提供接口實現類

package com.fs.springcloud.service.impl;import com.fs.springcloud.service.IMessageProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.messaging.Source; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Service;import java.util.UUID;//定義消息的推送管道 @Service @EnableBinding(Source.class)//指信道channel和Exchange綁定在一起 public class IMessageProviderImpl implements IMessageProvider {@Autowiredprivate MessageChannel output;//消息發送管道//重寫send方法,發送消息@Overridepublic String send() {String s = UUID.randomUUID().toString()+"我是provider使用StreamRabbitMQ發送的消息~~~";output.send(MessageBuilder.withPayload(s).build());System.out.println("~~~~消息"+s);return "success";} }

SendMessageController 服務提供方,用于調用send方法測試發送消息

package com.fs.springcloud.controller;import com.fs.springcloud.service.IMessageProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController;/* 提供方法,發送消息*/ @RestController public class SendMessageController {@Autowiredprivate IMessageProvider iMessageProvider;@RequestMapping("/sendMessage")public String sendMessage(){return iMessageProvider.send();} }

fs-stream-RabbitMQ-consumer-9901 Stream 消息消費者

創建消息消費者模塊，引入依賴 starter-stream-rabbit

編寫配置，定義 binder，和 bingings

定義消息接收業務類。添加 @EnableBinding(Sink.class)，使用
@StreamListener(Sink.INPUT)，完成消息接收。

編寫啟動類，測試

pom.xml

application.yml

server:port: 9901 spring:application:name: fs-stream-consumercloud:stream:binders: #在此處聲明配置要綁定的RabbitMQ的服務信息fs_Rabbit: # 表示定義的名稱,用binding整合type: rabbit # 消息組件類型environment: #設置RabbitMQ的相關配置spring:rabbitmq:host: 192.168.93.132port: 5672username: root #RabbitMQ的賬戶密碼password: rootvirtual-host: /bindings: #服務的整合處理input: # 這個名字是一個通道的名稱默認inputdefault-binder: fs_Rabbit # 設置要綁定的消息服務的具體設置, 爆紅沒關系destination: fs_Exchange #表示要使用的Exchange名稱定義content-type: application/json # 設置消息類型,本次為json 文本設置為test/plaingroup: xiaofuA #對消費者進行分組,將2個微服務分成同一個組,輪詢消費,解決重復消費#eureka: # client: # service-url: # defaultZone: http://fseureka7001.com:7001/eureka,http://fseureka7002.com:7002/eureka,http://fseureka7003.com:7003/eureka # instance: # lease-renewal-interval-in-seconds: 2 # 設置心跳的時間間隔(默認30秒) # lease-expiration-duration-in-seconds: 5 # 如果現在超過了5秒間隔(默認是90秒) # instance-id: receive-8803.com # 在消息列表顯示主機名稱 # prefer-ip-address: true # 訪問的路徑變ip地址

StreamRabbitMQMain9901 主啟動

package com.fs.springcloud;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; /* 故障現象,重復消費微服務應用放置于同一個group中,就能夠保證消息只會被其中一個應用消費一次不同的組是可以重復消費的,同一個租內會發生競爭關系,只有其中一個可以消費*/@SpringBootApplication public class StreamRabbitMQMain9901 {public static void main(String[] args) {SpringApplication.run(StreamRabbitMQMain9901.class,args);} }

ReceiveMessageListenerController 用于消費服務提供方發送的消息

package com.fs.springcloud.controller;import org.springframework.beans.factory.annotation.Value; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.messaging.Message; import org.springframework.stereotype.Component;/* 消息接收類*/ @Component @EnableBinding(Sink.class)//綁定消息 public class ReceiveMessageListenerController {@Value("${server.port}")private String serverPort;//定義消息接收方法@StreamListener(Sink.INPUT)public void input(Message<String> message){//打印消息數據//只要提供方發生了消息,這里就會被執行System.out.println("消費者,~~~~~收到的消息:"+message.getPayload()+",port:"+serverPort);} }

測試Stream 消息驅動小案列

首先打開RabbitMQ

啟動2個服務:
fs-stream-RabbitMQ-provider-8801
fs-stream-RabbitMQ-consumer-9901

瀏覽器發送服務提供方法的controller的地址調用send方法,發送消息到RabbitMQ中

然后觀看控9901的控制臺

Sleuth+Zipkin 鏈路追蹤

概述

? Spring Cloud Sleuth 其實是一個工具,它在整個分布式系統中能跟蹤一個用戶請求的過程，捕獲這些跟蹤數據，就能構建微服務的整個調用鏈的視圖，這是調試和監控微服務的關鍵工具。
? 耗時分析
? 可視化錯誤
? 鏈路優化
? Zipkin 是 Twitter 的一個開源項目，它致力于收集服務的定時數據，以解決微服務架構中的延遲問題，包括數據的收集、存儲、查找和展現。

定義：
- sleuth：采集程序中的請求等相關數據的工具
- zipkin：分析并圖形化展示調用關系的工具

搭建Sleuth+Zipkin項目

安裝啟動zipkin。 java –jar zipkin.jar

訪問zipkin web界面。 http://localhost:9411/

在服務提供方和消費方分別引入 sleuth 和 zipkin 依賴

分別配置服務提供方和消費方。

啟動，測試

使用：
- 服務端（ZipKin）
  
  jar包運行
  docker 運行
- 客戶端
  - 導包
    <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId> </dependency>
  - 配置
    - zipkin服務器地址信息
    - 采樣率
    spring:zipkin:base-url: http://localhost:9411/ # 設置zipkin的服務端路徑sleuth:sampler:probability: 1 # 采集率默認 0.1 百分之十。
  - 編碼
    
    無需任何編碼,服務自動被采集數據到zipkin

總結

以上是生活随笔為你收集整理的SpringCloud微服务架构,Config 分布式配置中心,Bus 消息总线, Stream 消息驱动,Sleuth+Zipkin 链路追踪的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： SpringCloud微服务架构之,Hy
下一篇： RabbitMQ,RabbitMQ 的工

javascript

SpringCloud微服务架构,Config 分布式配置中心,Bus 消息总线, Stream 消息驱动,Sleuth+Zipkin 链路追踪

Config分布式配置中心

Config 概述

Bus 消息總線

使用RabbitMQ做消息中間件

Bus 簡單項目案例(代碼中有詳細解釋)

搭建一個Springcloud的父項目

父項目 pom.xml

fs-server-eureka-7001(請參考另一篇博文,服務治理,有詳細說明搭建)

fs-config-server 搭建configserver

pom.xml

application.yml

ConfigServer3344 主啟動

fs-config-provider-8001/8002 搭建服務提供,來動態獲取配置信息

pom.xml

bootstrap.yml

application.yml空 沒有任何代碼

ConfigProvider8001/8002 主啟動

ConfigProviderController

. 啟動測試

在giree 碼云上準備一個廠庫(廠庫名字與git地址記得在configserver的application.yml文件中修改)

首先啟動RabbitMQ

啟動服務

打開瀏覽器測試

修改gitee上的配置文件信息后,再次打開瀏覽器測試

發送post請求廣播通知后,再次瀏覽器測試:

Stream 消息驅動

Stream 概述

Stream 組件

小案列演示 Stream 消息驅動

fs-stream-RabbitMQ-provider-8801 Stream 消息生產者

pom.xml

application.yml

StreamMain8801 主啟動

IMessageProvider 服務提供接口

IMessageProviderImpl 服務提供接口實現類

SendMessageController 服務提供方,用于調用send方法測試發送消息

fs-stream-RabbitMQ-consumer-9901 Stream 消息消費者

pom.xml

application.yml

StreamRabbitMQMain9901 主啟動

ReceiveMessageListenerController 用于消費服務提供方發送的消息

測試Stream 消息驅動 小案列

Sleuth+Zipkin 鏈路追蹤

概述

搭建Sleuth+Zipkin項目

總結

application.yml空沒有任何代碼

測試Stream 消息驅動小案列