[TOC] 前言 開發環境 部署環境:阿裡雲ECS伺服器 操作系統:CentOS 7.7 64位 IDEA 版本:2019.3.1 docker 版本:1.13.1 docker compose版本:1.25.0 spring cloud 版本:Hoxton.SR1 spring boot 版本:2 ...
目錄
- 前言
- 服務註冊中心(Eureka)
- 負載均衡(Ribbon)
- 聲明式服務調用(Feign)
- 統一配置中心(Config)
- 非同步消息(Stream)
- 微服務網關(Zuul)
- 服務容錯(Hystrix)
- 服務追蹤(Sleuth)
- 容器化部署
前言
開發環境
部署環境:阿裡雲ECS伺服器
- 操作系統:CentOS 7.7 64位
- IDEA 版本:2019.3.1
- docker 版本:1.13.1
- docker-compose版本:1.25.0
- spring cloud 版本:Hoxton.SR1
- spring boot 版本:2.2.2.RELEASE
- mysql 版本:5.7
- redis 版本:5.0.7
- rabbitmq 版本:3.8.2-management
- zipkin 版本:2.19.2
埠映射信息:
eureka1:8761 | eureka2:8762
config-server:8888
shopping-product:11100
shopping-order:11110
api-gateway:8080
open-api:8081
源碼地址
https://github.com/lizzie2008/spring-cloud-app.git
創建工程
- 新建一個工程:選擇Spring Cloud Bootstrap,對應的Spring Boot 版本2.2.2。
- 項目生成後,看到對應的Spring版本的依賴沒有問題。
- 因為是父工程,我們將打包格式改成pom,並把src等無用的文件刪除。
<groupId>tech.lancelot</groupId>
<artifactId>spring-cloud-app</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>spring-cloud-app</name>
<description>Demo project for Spring Cloud</description>
<packaging>pom</packaging>
服務註冊中心(Eureka)
Eureka Server
- 新建Module->選擇Eureka Server
因為Module作為子項目,我們改寫下對應的POM文件。
<parent> <groupId>tech.lancelot</groupId> <artifactId>spring-cloud-app</artifactId> <version>0.0.1-SNAPSHOT</version> </parent> <groupId>tech.lancelot</groupId> <artifactId>eureka-server</artifactId> <version>0.0.1-SNAPSHOT</version> <name>eureka-server</name> <description>Registry Center</description> <dependencies> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId> </dependency> </dependencies>
重新Build一下項目,能正常編譯。但是此時Eureka Server是不能正常啟動工作的,需要在application類增加
@EnableEurekaServer
。
此時,我們再運行Eureka Server,發現可以正常啟動服務註冊伺服器,服務埠8080,註冊地址:http://localhost:8761/eureka/。
- 打開瀏覽器,訪問8080埠,查看可視化管理界面。
- 當然,我們沒有做任何配置,並且控制台一直報錯,這是因為預設情況下,本身也是需要獲取註冊信息和註冊到註冊中心,而此時找不到對應伺服器。我們可以修改配置文件,做相應的配置。調整服務埠為8761,重新啟動後,發現不再報錯。
eureka:
client:
fetch-registry: false #設置不從註冊中心獲取註冊信息
register-with-eureka: false #設置自身不作為客戶端註冊到註冊中心
spring:
application:
name: eureka-server #應用名稱
server:
port: 8761 #應用服務埠
Eureka Client
- 我們再建一個Module工程,作為服務客戶端,向Eureka Server服務中心註冊。
同樣,我們修改POM文件,依賴於父項目,註意這裡需要引入
eureka-client
和spring-boot-starter-web
依賴。<parent> <groupId>tech.lancelot</groupId> <artifactId>spring-cloud-app</artifactId> <version>0.0.1-SNAPSHOT</version> </parent> <groupId>tech.lancelot</groupId> <artifactId>shopping-provider</artifactId> <version>0.0.1-SNAPSHOT</version> <name>shopping-provider</name> <description>shopping service provider</description> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> </dependency> </dependencies>
需要在application類增加
@EnableDiscoveryClient
,同時修改配置文件。eureka: client: serviceUrl: defaultZone: http://localhost:8761/eureka/ #指定服務註冊地址 spring: application: name: shopping-provider #應用名稱
重啟Eureka Client,啟動後再次訪問Eureka Server管理界面,可以發現order-provider服務已註冊。
註冊中心高可用
之前我們的Eureka Server是單點服務,實際生產中,經常是多台註冊中心,因此我們嘗試下配置2台註冊中心。
啟動伺服器實例1:
eureka: client: # fetch-registry: false #設置不從註冊中心獲取註冊信息 # register-with-eureka: false #設置自身不作為客戶端註冊到註冊中心 defaultZone: http://localhost:8762/eureka/ #指定服務註冊地址 spring: application: name: eureka-server1 #應用名稱 server: port: 8761 #應用服務埠
啟動伺服器實例2:
eureka: client: # fetch-registry: false #設置不從註冊中心獲取註冊信息 # register-with-eureka: false #設置自身不作為客戶端註冊到註冊中心 defaultZone: http://localhost:8761/eureka/ #指定服務註冊地址 spring: application: name: eureka-server2 #應用名稱 server: port: 8762 #應用服務埠
重啟2台註冊中心,啟動後分別訪問2台的管理界面,可以看到2台註冊中心已經相互註冊。
小結
項目增加2個服務模塊,並向Eureka Server註冊:shopping-product(商品服務)、shopping-order(訂單服務),實現相應業務邏輯,這部分詳細實現不再闡述。
整體項目結構如下:
spring-cloud-app
--eureka-server(服務註冊中心)
--shopping-common(購物公共模塊)
--shopping-product(商品服務模塊)
--shopping-order(訂單服務模塊)
系統架構如圖,比較簡單,一個集群服務中心,目前有2個服務提供並註冊:
負載均衡(Ribbon)
Spring Cloud Ribbon 是一個客戶端的負載均衡器,它提供對大量的HTTP和TCP客戶端的訪問控制。
客戶端負載均衡即是當瀏覽器向後臺發出請求的時候,客戶端會向 Eureka Server 讀取註冊到伺服器的可用服務信息列表,然後根據設定的負載均衡策略(沒有設置即用預設的),抉擇出向哪台伺服器發送請求。
假設有以下業務場景,shopping-order模塊需要調用shopping-product提供的API介面。我們看如何實現。
RestTemplate調用
第一種方法使用構造RestTemplate,調用遠程API,這種方法url是寫死,如果啟動多台shopping-product服務的話,那又該如何?
@Test
void getProductByRestTemplate() {
//1.第一種方法
RestTemplate restTemplate = new RestTemplate();
String response = restTemplate.getForObject("http://localhost:11100/api/products", String.class);
Assert.hasLength(response,"未獲取內容");
}
負載均衡調用
第二種方法:我們啟動2個shopping-product服務實例,分別是11100埠和9001埠,運行測試發現,會根據loadBalancerClient負載均衡機制幫我們選擇一個服務地址,進行訪問調用。
@Autowired
private LoadBalancerClient loadBalancerClient;
@Test
void getProductByLoadBalance(){
//2.第二種方法,先獲取負載均衡的地址再調用API
ServiceInstance instance = loadBalancerClient.choose("shopping-product");
String url=String.format("http://%s:%s/api/products",instance.getHost(),instance.getPort());
RestTemplate restTemplate = new RestTemplate();
String response = restTemplate.getForObject(url, String.class);
log.info("port:"+instance.getPort()+response);
}
應用名稱調用
但這樣依舊很是麻煩,接下來看第三種方法。第三種方法屏蔽了API的具體url信息,只用ServerId,並根據負載均衡規則,自動路由到對應的地址。
因為eureka包中已經添加了對Ribbon的依賴,我們可以增加斷點,調試程式,發現進入RibbonLoadBalancerClient-->choose方法,返回負載均衡策略選擇的ServiceInstance。
@Component
public class RestTemplateConfiguration {
@Bean
@LoadBalanced
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
@SpringBootTest
@Slf4j
class OrderServiceTest {
@Autowired
private RestTemplate restTemplate;
@Test
void getProductByServerId() {
String response = restTemplate.getForObject("http://shopping-product/api/products", String.class);
log.info(response);
}
}
當然,我們也可以指定應用服務的負載均衡策略:
shopping-order:
ribbon:
NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule
小結
目前系統架構如圖,實現shopping-product和shopping-order集群化部署,調用方式通過客戶端負載均衡,來路由消費端的請求。
聲明式服務調用(Feign)
Feign是一個聲明式的Web Service客戶端,它的目的就是讓Web Service調用更加簡單。Feign提供了HTTP請求的模板,通過編寫簡單的介面和插入註解,就可以定義好HTTP請求的參數、格式、地址等信息。
而Feign則會完全代理HTTP請求,我們只需要像調用方法一樣調用它就可以完成服務請求及相關處理。Feign整合了Ribbon和Hystrix(關於Hystrix我們後面再講),可以讓我們不再需要顯式地使用這兩個組件。
總起來說,Feign具有如下特性:
- 可插拔的註解支持,包括Feign註解和JAX-RS註解;
- 支持可插拔的HTTP編碼器和解碼器;
- 支持Hystrix和它的Fallback;
- 支持Ribbon的負載均衡;
- 支持HTTP請求和響應的壓縮。
服務端實現
shopping-product服務提供端暴露API。
@GetMapping("/productInfos")
public List<ProductInfoOutput> findProductInfosByIds(@RequestParam(required = false) String productIds) throws Exception {
//如果傳入商品id參數
if (StringUtils.isNotEmpty(productIds)) {
List<String> ids = Arrays.asList(productIds.split(","));
List<ProductInfo> productInfos = productService.findProductInfosByIds(ids);
List<ProductInfoOutput> productInfoOutputs = ListUtils.copyProperties(productInfos, ProductInfoOutput.class);
return productInfoOutputs;
}else{
List<ProductInfo> productInfos = productService.findProductInfos();
List<ProductInfoOutput> productInfoOutputs = ListUtils.copyProperties(productInfos, ProductInfoOutput.class);
return productInfoOutputs;
}
}
客戶端實現
- 引入Feign
shopping-order模塊需要調用shopping-product介面,首先我們在服務調用端增加Maven依賴
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-feign</artifactId>
</dependency>
啟動類標註開啟Feign服務
@SpringBootApplication
@EnableDiscoveryClient
@EnableFeignClients
public class ShoppingOrderApplication {
public static void main(String[] args) {
SpringApplication.run(ShoppingOrderApplication.class,args);
}
}
- 創建聲明式服務
/**
* 聲明式服務
*/
@FeignClient("shopping-product/api/v1")
public interface ProductClient {
@GetMapping("/productInfos")
List<ProductInfoOutput> findProductInfosByIds(@RequestParam(required = false) String productIds);
}
@FeignClient(“服務名稱”)映射服務調用,本質還是http請求,只不過Feign幫我們屏蔽了底層的請求路由,對開發者完全透明,使得調用遠程服務感覺跟調用本地服務一致的編碼體驗。
本地調用測試,可以正常返回介面數據。
@GetMapping("/orders/findProductInfosByIds")
public List<ProductInfoOutput> findProductInfosByIds(){
List<ProductInfoOutput> productInfoOutputs = productClient
.findProductInfosByIds("157875196366160022, 157875227953464068");
return productInfoOutputs;
}
小結
在實現負載均衡基礎上,封裝聲明式服務調用。實現shopping-order對shopping-product的透明調用,系統架構如圖如下。
統一配置中心(Config)
上個環境中,我們有2個服務提供者,首先看下各自的配置,可以發現很大一部分都是重覆的。
如果微服務架構中沒有使用統一配置中心時,所存在的問題:
- 配置文件分散在各個項目里,不方便維護
- 配置內容安全與許可權,實際開發中,開發人員是不知道線上環境的配置的
- 更新配置後,項目需要重啟
eureka:
client:
serviceUrl:
defaultZone: http://localhost:8761/eureka/ #指定服務註冊地址
spring:
application:
name: shopping-order #應用名稱
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
username: root
password: 123456
url: jdbc:mysql://localhost:3306/spring_cloud_app?characterEncoding=utf-8&useSSL=false&serverTimezone=UTC
jpa:
show-sql: true
database-platform: org.hibernate.dialect.MySQLDialect
server:
port: 11110
對於一些簡單的項目來說,我們一般都是直接把相關配置放在單獨的配置文件中,以 properties 或者 yml 的格式出現,更省事兒的方式是直接放到 application.properties 或 application.yml 中。在集群部署情況下,我們嘗試來實現配置的集中管理,並支持配置的動態刷新。
Config Server
- 我們新建一個Module工程,統一配置中心,保存所以的配置信息。
同樣,我們作為子項目,修改相關依賴,加入對spring-cloud-config-server依賴
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>tech.lancelot</groupId>
<artifactId>spring-cloud-app</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<artifactId>config-server</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- spring cloud config 服務端包 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-config-server</artifactId>
</dependency>
</dependencies>
- application.properties進行如下配置
spring:
application:
name: config-server # 應用名稱
cloud:
config:
server:
git:
uri: https://github.com/lizzie2008/Central-Configuration.git #配置文件所在倉庫
username: 'Github username'
password: 'Github password'
default-label: master #配置文件分支
search-paths: spring-cloud-app #配置文件所在根目錄
server:
port: 8888
- 在 Application 啟動類上增加相關註解
@EnableConfigServer
@EnableConfigServer
@SpringBootApplication
public class ConfigServerApplication {
public static void main(String[] args) {
SpringApplication.run(ConfigServerApplication.class, args);
}
}
- 啟動服務,接下來測試一下。
Spring Cloud Config 有它的一套訪問規則,我們通過這套規則在瀏覽器上直接訪問就可以。
/{application}/{profile}[/{label}]
/{application}-{profile}.yml
/{label}/{application}-{profile}.yml
/{application}-{profile}.properties
/{label}/{application}-{profile}.properties
{application} 就是應用名稱,對應到配置文件上來,就是配置文件的名稱部分,例如我上面創建的配置文件。
{profile} 就是配置文件的版本,我們的項目有開發版本、測試環境版本、生產環境版本,對應到配置文件上來就是以 application-{profile}.yml 加以區分,例如application-dev.yml、application-sit.yml、application-prod.yml。
{label} 表示 git 分支,預設是 master 分支,如果項目是以分支做區分也是可以的,那就可以通過不同的 label 來控制訪問不同的配置文件了。
我們在git項目中,新建spring-cloud-app/config-eureka-server.yml配置文件,然後訪問配置中心伺服器,看看能正常獲取配置文件。
向服務中心註冊
config-server本身作為一個服務,也可以作為服務提供方,向服務中心註冊,其他的服務想要獲取配置文件,只需要通過服務名稱就會訪問。
- 引入Eureka Client依賴
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
- 啟動類上增加
@EnableDiscoveryClient
註解
@EnableConfigServer
@EnableDiscoveryClient
@SpringBootApplication
public class ConfigServerApplication {
public static void main(String[] args) {
SpringApplication.run(ConfigServerApplication.class, args);
}
}
- 配置文件中增加eureka註冊。
eureka:
client:
serviceUrl:
defaultZone: http://eureka1:8761/eureka/,http://eureka2:8762/eureka/ #指定服務註冊地址
- 啟動eureka-server,看看config-server是否註冊成功。
服務提供端改造
- shopping-product項目中,把原先的application.yml文件重命名為bootstrap.yml,並配置Eureka Server地址、應用名稱、Config的實例名稱。服務啟動後,會鏈接Eureka Server伺服器,根據Config的實例名稱找到對應的Config伺服器,並根據實例名稱(可以增加profile屬性)來匹配配置文件。
eureka:
client:
serviceUrl:
defaultZone: http://eureka1:8761/eureka/,http://eureka2:8762/eureka/ #指定服務註冊地址
spring:
application:
name: shopping-product #應用名稱
cloud:
config:
discovery:
enabled: true
service-id: config-server
- 之前服務端其餘的配置,填寫在github配置項目shopping-product.yml文件中
spring:
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
username: root
password: 123456
url: jdbc:mysql://localhost:3306/spring_cloud_app?characterEncoding=utf-8&useSSL=false&serverTimezone=UTC
jpa:
show-sql: true
database-platform: org.hibernate.dialect.MySQLDialect
server:
port: 11100
- 同樣,shopping-order項目也如此改造,最後我們啟動所有的服務,看是否都能正常啟動。
配置動態刷新
- 首先,在
shopping-product.yml
增加一個配置屬性來進行測試
env: dev
- 新建一個測試controller來綁定這個配置屬性,並提供api來返回屬性的值
@RestController
@RefreshScope
@RequestMapping("api/env")
public class EnvController {
@Value("${env}")
private String env;
@RequestMapping
public String printEnv() {
return env;
}
}
訪問http://localhost:11100/api/env,返回當前的值dev。
Spring Cloud Config 在項目啟動時載入配置內容這一機制,但是如果我們修改配置文件內容後,不會自動刷新。例如我們上面的項目,當服務已經啟動的時候,去修改 github 上的配置文件內容,這時候,再次刷新頁面,對不起,還是舊的配置內容,新內容不會主動刷新過來。那應該怎麼去觸發配置信息的動態刷新呢?
它提供了一個刷新機制,但是需要我們主動觸發。那就是 @RefreshScope 註解並結合 actuator ,註意要引入 spring-boot-starter-actuator 包。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
- EnvController上增加
@RefreshScope
註解 - 發送 POST 請求到 http://localhost:11100/actuator/refresh 這個介面,預設沒有開放endpoint的許可權,所以這塊我們首先配置開放許可權
management:
endpoints:
web:
exposure:
include: "*"
- 這時調用介面結束後,我們看到介面返回消息,表明env這個屬性值已經刷新
[
"config.client.version",
"env"
]
- 再次訪問http://localhost:11100/api/env,返回當前的值就是修改後的值test,證明配置屬性的值已經動態刷新,我們的程式也不用再次啟動。
配置 Webhook
每次改了配置後,就用 postman 訪問一下 refresh 介面,還是不夠方便。 github 提供了一種 webhook 的方式,當有代碼變更的時候,會調用我們設置的地址,來實現我們想達到的目的。
- 進入 github 倉庫配置頁面,選擇 Webhooks ,並點擊 add webhook;
填上回調的地址
也就是上面提到的 actuator/refresh 這個地址,但是必須保證這個地址是可以被 github 訪問到的。這樣每當github上修改了配置文件,就自動通知對應的hook地址自動刷新。
小結
整體項目結構如下:
spring-cloud-app
--config-server(統一配置中心)
--eureka-server(服務註冊中心)
--shopping-common(購物公共模塊)
--shopping-product(商品服務模塊)
--shopping-order(訂單服務模塊)
更新系統架構,新建config-server節點,也向eureka-server註冊,相關服務註冊節點根據配置實例名稱,路由到config-server節點,動態的載入配置。
非同步消息(Stream)
應用場景
1、非同步處理
比如用戶在電商網站下單,下單完成後會給用戶推送簡訊或郵件,發簡訊和郵件的過程就可以非同步完成。因為下單付款是核心業務,發郵件和簡訊並不屬於核心功能,並且可能耗時較長,所以針對這種業務場景可以選擇先放到消息隊列中,有其他服務來非同步處理。
2、應用解耦:
假設公司有幾個不同的系統,各系統在某些業務有聯動關係,比如 A 系統完成了某些操作,需要觸發 B 系統及 C 系統。如果 A 系統完成操作,主動調用 B 系統的介面或 C 系統的介面,可以完成功能,但是各個系統之間就產生了耦合。用消息中間件就可以完成解耦,當 A 系統完成操作將數據放進消息隊列,B 和 C 系統去訂閱消息就可以了。這樣各系統只要約定好消息的格式就好了。
3、流量削峰
比如秒殺活動,一下子進來好多請求,有的服務可能承受不住瞬時高併發而崩潰,所以針對這種瞬時高併發的場景,在中間加一層消息隊列,把請求先入隊列,然後再把隊列中的請求平滑的推送給服務,或者讓服務去隊列拉取。
4、日誌處理
kafka 最開始就是專門為了處理日誌產生的。
當碰到上面的幾種情況的時候,就要考慮用消息隊列了。如果你碰巧使用的是 RabbitMQ 或者 kafka ,而且同樣也是在使用 Spring Cloud ,那可以考慮下用 Spring Cloud Stream。Spring Cloud Stream 是消息中間件組件,它集成了 kafka 和 rabbitmq ,本文以rabbitmq 為例。
當前項目場景
分析目前shopping-order項目中,創建訂單的代碼如下:
/**
* 創建訂單
*
*/
@Transactional
public String Create(OrderInput orderInput) throws Exception {
//扣庫存
ResultVo result1=productClient.decreaseStock(orderInput.getOrderItemInputs());
if (result1.getCode() != 0)
throw new Exception("調用訂單扣減庫存介面出錯:" + result1.getMsg());
//構建訂單主表
OrderMaster orderMaster = new OrderMaster();
BeanUtils.copyProperties(orderInput, orderMaster);
//指定預設值
orderMaster.setOrderId(KeyUtil.genUniqueKey("OM"));
orderMaster.setOrderStatus(OrderStatus.NEW);
orderMaster.setPayStatus(PayStatus.WAIT);
//構建訂單明細
List<String> productIds = orderInput.getOrderItemInputs().stream().map(OrderItemInput::getProductId).collect(Collectors.toList());
ResultVo<List<ProductInfoOutput>> result2 = productClient.findProductInfosByIds(String.join(",", productIds));
if (result2.getCode() != 0)
throw new Exception("調用訂單查詢介面出錯:" + result2.getMsg());
List<ProductInfoOutput> productInfoOutputs = result2.getData();
//訂單金額總計
BigDecimal total = new BigDecimal(BigInteger.ZERO);
for (OrderItemInput orderItemInput : orderInput.getOrderItemInputs()) {
OrderDetail orderDetail = new OrderDetail();
BeanUtils.copyProperties(orderItemInput, orderDetail);
Optional<ProductInfoOutput> productInfoOutputOptional = productInfoOutputs.stream()
.filter(s -> s.getProductId().equals(orderItemInput.getProductId())).findFirst();
if (!productInfoOutputOptional.isPresent())
throw new Exception(String.format("商品【%s】不存在", orderItemInput.getProductId()));
ProductInfoOutput productInfoOutput = productInfoOutputOptional.get();
orderDetail.setDetailId(KeyUtil.genUniqueKey("OD"));
orderDetail.setOrderId(orderMaster.getOrderId());
orderDetail.setProductName(productInfoOutput.getProductName());
orderDetail.setProductPrice(productInfoOutput.getProductPrice().multiply(new BigDecimal(orderDetail.getProductQuantity())));
orderDetail.setProductIcon(productInfoOutput.getProductIcon());
total = total.add(orderDetail.getProductPrice());
orderDetailRepository.save(orderDetail);
}
orderMaster.setOrderAmount(total);
orderMasterRepository.save(orderMaster);
return orderMaster.getOrderId();
}
創建訂單的同時,先調用商品介面扣減庫存,如果占用庫存成功,再生成訂單。這樣的話,生成訂單的操作和占用商品庫存的操作其實是耦合在一起的。在實際電商高併發、高流量的情況下,我們很少這麼做。所以,我們要將業務解耦,實現訂單和扣減庫存的非同步處理。
大體思路如下:生成訂單==》通知商品調用庫存==》商品占用庫存==》通知訂單占用成功==》更新訂單占用庫存狀態
stream-rabbit集成
shopping-order、shopping-product項目中
- 首先引入stream-rabbit依賴:
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
- application.yml中作相應的配置:
spring:
rabbitmq:
host: aliyun.host
port: 5672
username: guest
password: guest
- 消息介面定義
public interface StreamClient {
String INPUT = "myMessage";
@Input(StreamClient.INPUT)
SubscribableChannel input();
@Output(StreamClient.INPUT)
MessageChannel output();
}
- 接收端處理邏輯
@Component
@EnableBinding(StreamClient.class)
@Slf4j
public class StreamReceiver {
@StreamListener(value = StreamClient.INPUT)
public void process(OrderInput orderInput) {
log.info("StreamReceiver: {}", orderInput);
}
}
- 發送端處理邏輯
@RestController
@RequestMapping("api/v1/stream")
@Slf4j
public class StreamController {
private final StreamClient streamClient;
@Autowired
public StreamController(StreamClient streamClient) {
this.streamClient = streamClient;
}
@GetMapping("/sendMessage")
public void sendMessage() {
OrderInput orderInput=new OrderInput();
orderInput.setBuyerName("小王");
orderInput.setBuyerPhone("15011111111");
orderInput.setBuyerAddress("姥姥家");
orderInput.setBuyerOpenid("11111");
streamClient.output().send(MessageBuilder.withPayload(orderInput).build());
}
}
啟動應用程式,測試發送介面,發現spring-cloud-stream幫我們自動創建了一個隊列,消息發送到這個隊列,然後被接收端消費。
此時,如果我們啟動多個shopping-product服務實例,會有個問題,如果發送端發送一條消息,會被2個實例同時消費,在正常的業務中,這種情況是應該避免的。所以我們需要對消息進行分組,在application.yml中增加如下配置,保證只有一個服務實例來消費。
spring:
rabbitmq:
host: aliyun.host
port: 5672
username: guest
password: guest
cloud:
stream:
bindings:
myMessage:
group: shopping-order
content-type: application/json
改造Order和Product項目
shopping-order作為庫存占用命令的消息發送者,首先向shopping-product發送消息stock_apply(占用庫存申請),shopping-product接收此消息進行庫存處理,然後將庫存占用處理的結果作為消息stock_result(占用庫存結果)發送,shopping-order端再收到結果消息對訂單狀態進行更新。
- shopping-order配置:
spring:
cloud:
stream:
bindings:
stock_apply_output: #占用庫存申請
destination: stock.apply
stock_result_input: #占用庫存結果
destination: stock.result
group: shopping-order
- shopping-product配置:
spring:
cloud:
stream:
bindings:
stock_apply_input: #占用庫存申請
destination: stock.apply
group: shopping-product
stock_result_output: #占用庫存結果
destination: stock.result
- shopping-order定義channel
public interface OrderStream {
String STOCK_APPLY_OUTPUT = "stock_apply_output";
@Output(OrderStream.STOCK_APPLY_OUTPUT)
MessageChannel stockApplyOutput();
String STOCK_RESULT_INPUT = "stock_result_input";
@Input(OrderStream.STOCK_RESULT_INPUT)
SubscribableChannel stockResultInput();
}
- shopping-product定義channel
public interface ProductStream {
String STOCK_APPLY_INPUT = "stock_apply_input";
@Input(ProductStream.STOCK_APPLY_INPUT)
SubscribableChannel stockApplyInput();
String STOCK_RESULT_OUTPUT = "stock_result_output";
@Output(ProductStream.STOCK_RESULT_OUTPUT)
MessageChannel stockResultOutput();
}
- shopping-order發送庫存申請消息
/**
* 創建訂單
*/
@Transactional
public String Create(OrderInput orderInput) throws Exception {
//構建訂單主表
OrderMaster orderMaster = new OrderMaster();
BeanUtils.copyProperties(orderInput, orderMaster);
//指定預設值
orderMaster.setOrderId(KeyUtil.genUniqueKey("OM"));
orderMaster.setOrderStatus(OrderStatus.NEW);
orderMaster.setPayStatus(PayStatus.WAIT);
//構建訂單明細
List<String> productIds = orderInput.getOrderItemInputs().stream().map(OrderItemInput::getProductId).collect(Collectors.toList());
ResultVo<List<ProductInfoOutput>> result2 = productClient.findProductInfosByIds(String.join(",", productIds));
if (result2.getCode() != 0)
throw new Exception("調用訂單查詢介面出錯:" + result2.getMsg());
List<ProductInfoOutput> productInfoOutputs = result2.getData();
//訂單金額總計
BigDecimal total = new BigDecimal(BigInteger.ZERO);
for (OrderItemInput orderItemInput : orderInput.getOrderItemInputs()) {
OrderDetail orderDetail = new OrderDetail();
BeanUtils.copyProperties(orderItemInput, orderDetail);
Optional<ProductInfoOutput> productInfoOutputOptional = productInfoOutputs.stream()
.filter(s -> s.getProductId().equals(orderItemInput.getProductId())).findFirst();
if (!productInfoOutputOptional.isPresent())
throw new Exception(String.format("商品【%s】不存在", orderItemInput.getProductId()));
ProductInfoOutput productInfoOutput = productInfoOutputOptional.get();
orderDetail.setDetailId(KeyUtil.genUniqueKey("OD"));
orderDetail.setOrderId(orderMaster.getOrderId());
orderDetail.setProductName(productInfoOutput.getProductName());
orderDetail.setProductPrice(productInfoOutput.getProductPrice().multiply(new BigDecimal(orderDetail.getProductQuantity())));
orderDetail.setProductIcon(productInfoOutput.getProductIcon());
total = total.add(orderDetail.getProductPrice());
orderDetailRepository.save(orderDetail);
}
orderMaster.setOrderAmount(total);
orderMasterRepository.save(orderMaster);
//扣庫存
StockApplyInput stockApplyInput = new StockApplyInput();
stockApplyInput.setOrderId(orderMaster.getOrderId());
stockApplyInput.setOrderItemInputs(orderInput.getOrderItemInputs());
orderStream.stockApplyOutput().send(MessageBuilder.withPayload(stockApplyInput).build());
return orderMaster.getOrderId();
}
- shopping-product處理庫存申請消息,併發送庫存處理結果
@Service
@Slf4j
@EnableBinding(ProductStream.class)
public class ProductService {
private final ProductInfoRepository productInfoRepository;
private final ProductCategoryRepository productCategoryRepository;
@Autowired
public ProductService(ProductInfoRepository productInfoRepository,
ProductCategoryRepository productCategoryRepository) {
this.productInfoRepository = productInfoRepository;
this.productCategoryRepository = productCategoryRepository;
}
/**
* 扣減庫存
*
*/
@Transactional
@StreamListener(ProductStream.STOCK_APPLY_INPUT)
@SendTo(ProductStream.STOCK_RESULT_OUTPUT)
public StockResultOutput processStockApply(StockApplyInput stockApplyInput) throws Exception {
log.info("占用庫存消息被消費...");
StockResultOutput stockResultOutput = new StockResultOutput();
stockResultOutput.setOrderId(stockApplyInput.getOrderId());
try {
for (OrderItemInput orderItemInput : stockApplyInput.getOrderItemInputs()) {
Optional<ProductInfo> productInfoOptional = productInfoRepository.findById(orderItemInput.getProductId());
if (!productInfoOptional.isPresent())
throw new Exception("商品不存在.");
ProductInfo productInfo = productInfoOptional.get();
int result = productInfo.getProductStock() - orderItemInput.getProductQuantity();
if (result < 0)
throw new Exception("商品庫存不滿足.");
productInfo.setProductStock(result);
productInfoRepository.save(productInfo);
}
stockResultOutput.setIsSuccess(true);
stockResultOutput.setMessage("OK");
return stockResultOutput;
} catch (Exception e) {
stockResultOutput.setIsSuccess(false);
stockResultOutput.setMessage(e.getMessage());
return stockResultOutput;
}
}
}
- shopping-order處理庫存處理結果
@StreamListener(OrderStream.STOCK_RESULT_INPUT)
public void processStockResult(StockResultOutput stockResultOutput) {
log.info("庫存消息返回" + stockResultOutput);
Optional<OrderMaster> optionalOrderMaster = orderMasterRepository.findById(stockResultOutput.getOrderId());
if (optionalOrderMaster.isPresent()) {
OrderMaster orderMaster = optionalOrderMaster.get();
if (stockResultOutput.getIsSuccess()) {
orderMaster.setOrderStatus(OrderStatus.OCCUPY_SUCCESS);
} else {
orderMaster.setOrderStatus(OrderStatus.OCCUPY_FAILURE);
}
orderMasterRepository.save(orderMaster);
}
}
執行調試結果,跟蹤執行結果:生成訂單同時發送庫存申請命令,商品模塊處理庫存申請成功後,返回庫存占用結果告知訂單模塊,從而實現訂單生成和商品庫存占用的邏輯的解耦。
小結
在原有的架構基礎上,我們對商品和訂單服務進行了應用解耦,庫存占用邏輯非同步化,通過消息隊列傳遞消息,並結合spring cloud stream對消息input和output綁定,使得在程式中很方便的進行消息發送和接收處理。
微服務網關(Zuul)
Zuul是Netflix開源的微服務網關,可以和Eureka、Ribbon、Hystrix等組件配合使用,Spring Cloud對Zuul進行了整合與增強,Zuul預設使用的HTTP客戶端是Apache HTTPClient,也可以使用RestClient或okhttp3.OkHttpClient。 Zuul的主要功能是路由轉發和過濾器。zuul預設和Ribbon結合實現了負載均衡的功能
工作原理
zuul的核心是一系列的filters, 其作用類比Servlet框架的Filter,或者AOP。zuul把請求路由到用戶處理邏輯的過程中,這些filter參與一些過濾處理,比如Authentication,Load Shedding等
Zuul使用一系列不同類型的過濾器,使我們能夠快速靈活地將功能應用於我們的邊緣服務。這些過濾器可幫助我們執行以下功能:
- 身份驗證和安全性 - 確定每個資源的身份驗證要求並拒絕不滿足這些要求的請求
- 洞察和監控 - 在邊緣跟蹤有意義的數據和統計數據,以便為我們提供準確的生產視圖
- 動態路由 - 根據需要動態地將請求路由到不同的後端群集
- 壓力測試 - 逐漸增加群集的流量以衡量性能。
- Load Shedding - 為每種類型的請求分配容量並刪除超過限制的請求
- 靜態響應處理 - 直接在邊緣構建一些響應,而不是將它們轉發到內部集群
添加網關
- 新建api-gateway子模塊,作為服務網關、服務發現客戶端、獲取配置客戶端,因此需要引入以下依賴。
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-config-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-zuul</artifactId>
</dependency>
</dependencies>
- 在啟動類上增加
EnableDiscoveryClient
和@EnableZuulProxy
註解。
@EnableDiscoveryClient
@EnableZuulProxy
@SpringBootApplication
public class ApiGatewayApplication {
public static void main(String[] args) {
SpringApplication.run(ApiGatewayApplication.class, args);
}
}
- 啟動服務,看看是否能正常獲取配置,並註冊到Eureka Server。
- Zuul網關目前暴露的埠是8080,之前我們訪問商品服務的api,是通過調用 http://localhost:11100/api/v1/product/productInfos來訪問的,現在我們就可以通過Zuul,根據商品的服務名稱shopping-produc來訪問 http://localhost:8080/shopping-product//api/v1/product/productInfos,非常輕鬆的實現了路由的功能。
自定義路由
預設的路由規則是按照服務的名稱來路由服務,當然我們也可以自定義。在zuul中,路由匹配的路徑表達式採用ant風格定義
通配符 | 說明 |
---|---|
? | 匹配任意單個字元 |
* | 匹配任意數量的字元 |
** | 匹配任意數量的字元,支持多級目錄 |
zuul:
routes:
# 簡潔寫法
shopping-product: /product/**
- 將命名為product的所有路徑都映射到shopping-product服務中去,然後通過product名稱來訪問,依舊能訪問成功。
- 如果我們需要查看目前所有的路徑映射呢,首先得引入actuator
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
- 其次,需要放開actuator維護埠的許可權
management:
endpoints:
web:
exposure:
include: "*"
- 訪問 http://localhost:8080/actuator/routes ,可以看到目前網關的所有路由映射
- 如果需要定義哪些方法不能通過網關調用,還可以設置排除哪些路由的規則
zuul:
routes:
# 簡潔寫法
shopping-product: /product/**
# 排除某些路由
ignored-patterns:
- /**/productInfos
這樣我們再訪問這個介面時,就提示 Not Found 錯誤了
Cookie與頭信息
預設情況下,spring cloud zuul在請求路由時,會過濾掉http請求頭信息中一些敏感信息,防止它們被傳遞到下游的外部伺服器。預設的敏感頭信息通過zuul.sensitiveHeaders參數定義,預設包括cookie,set-Cookie,authorization三個屬性。所以,我們在開發web項目時常用的cookie在spring cloud zuul網關中預設時不傳遞的,這就會引發一個常見的問題,如果我們要將使用了spring security,shiro等安全框架構建的web應用通過spring cloud zuul構建的網關來進行路由時,由於cookie信息無法傳遞,我們的web應用將無法實現登錄和鑒權。有時候,針對某些路由,我們需要傳遞這個cookie。
zuul:
routes:
# 完全寫法
product-route:
path: /product/**
serviceId: shopping-product
# 將指定路由的敏感頭設置為空
sensitiveHeaders:
動態路由
之前路由的配置都是寫在配置文件中,如果路由規則變化以後,需要重啟網關服務。但是實際生產環境,一般都需要動態的載入路由的配置,不能輕易重啟網關服務。
- 將配置信息集中到統一配置中心服務進行管理,具體實施參考前面章節-統一配置中心。
eureka:
client:
serviceUrl:
defaultZone: http://eureka1:8761/eureka/,http://eureka2:8762/eureka/ #指定服務註冊地址
spring:
application:
name: api-gateway #應用名稱
cloud:
config:
discovery:
enabled: true
service-id: config-server
- 將zuul配置屬性定義成支持動態刷新,增加
@RefreshScope
註解
@Component
public class ZuulConfiguration {
@ConfigurationProperties("zuul")
@RefreshScope
public ZuulProperties zuulProperties(){
return new ZuulProperties();
}
}
自定義Filter
設想以下場景:我們需要判斷用戶請求的參數是否包含認證信息,如果包含token信息,則可以訪問,否則禁止訪問。可以用Zuul Filter很方便的實現在網關端,統一進行認證。
- 新建自定義的Filter,並繼承ZuulFilter,預設需要實現4個介面
- filterType():返回 filter 的類型,設置為
PRE_TYPE
- filterOrder():返回 filter 的順序,設置為
PRE_DECORATION_FILTER_ORDER-1
- shouldFilter():是否啟用 filter,設置為
true
- run():執行具體的過濾器邏輯
- filterType():返回 filter 的類型,設置為
/**
* 驗證token 過濾器
*/
@Component
public class TokenFilter extends ZuulFilter {
@Override
public String filterType() {
return PRE_TYPE;
}
@Override
public int filterOrder() {
return 0;
}
@Override
public boolean shouldFilter() {
return true;
}
@Override
public Object run() throws ZuulException {
RequestContext currentContext = RequestContext.getCurrentContext();
HttpServletRequest request = currentContext.getRequest();
//測試在url參數中獲取token
String token = request.getParameter("token");
if(StringUtils.isEmpty(token)){
currentContext.setSendZuulResponse(false);
currentContext.setResponseStatusCode(HttpStatus.UNAUTHORIZED.value());
}
return null;
}
}
- 驗證結果,如果url中添加了 token 參數,TokenFilter 驗證通過,正確返回結果;如果沒有 token 參數,則返回 401(UNAUTHORIZED)錯誤
- 還可以在調用介面返回中,設置響應頭
@Component
public class AddResHeaderFilter extends ZuulFilter{
@Override
public String filterType() {
return POST_TYPE;
}
@Override
public int filterOrder() {
return SEND_RESPONSE_FILTER_ORDER - 1;
}
@Override
public boolean shouldFilter() {
return true;
}
@Override
public Object run() {
RequestContext requestContext = RequestContext.getCurrentContext();
HttpServletResponse response = requestContext.getResponse();
response.setHeader("X-Foo", UUID.randomUUID().toString());
return null;
}
}
限流
這裡介紹一種限流的設計方案:
對於很多應用場景來說,除了要求能夠限制數據的平均傳輸速率外,還要求允許某種程度的突發傳輸。這時候漏桶演算法可能就不合適了,令牌桶演算法更為適合。如圖所示,令牌桶演算法的原理是系統會以一個恆定的速度往桶里放入令牌,而如果請求需要被處理,則需要先從桶里獲取一個令牌,當桶里沒有令牌可取時,則拒絕服務。
Google公司已經實現了上述的令牌桶的演算法,直接使用 RateLimiter 就可以通過Zuul實現限流的功能:
@Component
public class RateLimitFilter extends ZuulFilter {
private static final RateLimiter RATE_LIMITER = RateLimiter.create(100);
@Override
public String filterType() {
return PRE_TYPE;
}
@Override
public int filterOrder() {
return SERVLET_DETECTION_FILTER_ORDER - 1;
}
@Override
public boolean shouldFilter() {
return true;
}
@Override
public Object run() {
if (!RATE_LIMITER.tryAcquire()) {
throw new RuntimeException("未能獲取到令牌.");
}
return null;
}
}
小結
整體項目結構如下:
spring-cloud-app
--api-gateway(服務網關)
--config-server(統一配置中心)
--eureka-server(服務註冊中心)
--shopping-common(購物公共模塊)
--shopping-product(商品服務模塊)
--shopping-order(訂單服務模塊)
目前所有的客戶端請求,首先被髮送到統一網關服務處理,然後由網關進行限流、熔斷、許可權驗證、記錄日誌等等,然後根據自定義的路由規則,再分發到不同的應用服務中去,應用伺服器返回處理結果後,由網關統一返回給客戶端。
服務容錯(Hystrix)
在分散式環境中,許多服務依賴項中的一些必然會失敗。Hystrix是一個庫,通過添加延遲容忍和容錯邏輯,幫助你控制這些分散式服務之間的交互。Hystrix通過隔離服務之間的訪問點、停止級聯失敗和提供回退選項來實現這一點,所有這些都可以提高系統的整體彈性。
設計原則
- 防止任何單個依賴項耗盡所有容器(如Tomcat)用戶線程。
- 甩掉包袱,快速失敗而不是排隊。
- 在任何可行的地方提供回退,以保護用戶不受失敗的影響。
- 使用隔離技術(如隔離板、泳道和斷路器模式)來限制任何一個依賴項的影響。
- 通過近實時的度量、監視和警報來優化發現時間。
- 通過配置的低延遲傳播來優化恢復時間。
- 支持對Hystrix的大多數方面的動態屬性更改,允許使用低延遲反饋迴圈進行實時操作修改。
- 避免在整個依賴客戶端執行中出現故障,而不僅僅是在網路流量中。
如何實現
- 用一個HystrixCommand 或者 HystrixObservableCommand (這是命令模式的一個例子)包裝所有的對外部系統(或者依賴)的調用,典型地它們在一個單獨的線程中執行
- 調用超時時間比你自己定義的閾值要長。有一個預設值,對於大多數的依賴項你是可以自定義超時時間的。
- 為每個依賴項維護一個小的線程池(或信號量);如果線程池滿了,那麼該依賴性將會立即拒絕請求,而不是排隊。
- 調用的結果有這麼幾種:成功、失敗(客戶端拋出異常)、超時、拒絕。
- 在一段時間內,如果服務的錯誤百分比超過了一個閾值,就會觸發一個斷路器來停止對特定服務的所有請求,無論是手動的還是自動的。
- 當請求失敗、被拒絕、超時或短路時,執行回退邏輯。
- 近實時監控指標和配置變化。
觸發降級
在實際工作中,尤其是分散式、微服務越來越普遍的今天,一個服務經常需要調用其他的服務,即RPC調用,而調用最多的方式還是通過http請求進行調用,這裡面就有一個問題了,如果調用過程中,因為網路等原因,造成某個服務調用超時,如果沒有熔斷機制,此處的調用鏈路將會一直阻塞在這裡,在高併發的環境下,如果許多個請求都卡在這裡的話,伺服器不得不為此分配更多的線程來處理源源不斷涌入的請求。
更恐怖的是,如果這是一個多級調用,即此處的服務的調用結果還被其他服務調用了,這就形成了所謂的雪崩效應,後果將不堪設想。因此,需要某種機制,在一定的異常介面調用出現的時候,能夠自動發現這種異常,並快速進行服務降級。
- 首先,shopping-order項目模擬一個遠程調用shopping-product服務http請求
/**
* Hystrix 測試
*/
@RestController
@RequestMapping("api/hystrix")
public class HystrixController {
@GetMapping("/getProductEnv")
public String getProductEnv() {
RestTemplate restTemplate = new RestTemplate();
return restTemplate.postForObject("http://localhost:11100/api/env", null, String.class);
}
}
- 如果此時將shopping-product服務關閉,則shopping-order調用遠程服務不可用,進入等待,超時時返回 Error Page的錯誤頁面。其實我們希望服務不可用的時候直接處理,返回通知服務的不可用狀態。可以引入Hystrix。
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
</dependency>
- 在啟動類上增加
@EnableCircuitBreaker
註解,或者將@SpringBootApplication
、@EnableDiscoveryClient