javascript
SpringCloud笔记（三）微服务应用
微服務應用
前面我們已經完成了SpringCloudAlibaba的學習，我們對一個微服務項目的架構體系已經有了一定的了解，那么本章我們將在應用層面繼續探討微服務。
分布式權限校驗
雖然完成前面的部分，我們已經可以自己去編寫一個比較中規中矩的微服務項目了，但是還有一個問題我們沒有解決，登錄問題。假如現在要求用戶登錄之后，才能進行圖書的查詢、借閱等操作，那么我們又該如何設計這個系統呢？
回顧我們之前進行權限校驗的原理，服務器是如何判定一個請求是來自哪個用戶的呢？
- 首先瀏覽器會向服務端發送請求，訪問我們的網站。
- 服務端收到請求后，會創建一個SESSION ID，并暫時存儲在服務端，然后會發送給瀏覽器作為Cookie保存。
- 之后瀏覽器會一直攜帶此Cookie訪問服務器，這樣在收到請求后，就能根據攜帶的Cookie中的SESSION ID判斷是哪個用戶了。
- 這樣服務端和瀏覽器之間可以輕松地建立會話了。
但是我們想一下，我們現在采用的是分布式的系統，那么在用戶服務進行登錄之后，其他服務比如圖書服務和借閱服務，它們會知道用戶登錄了嗎？
實際上我們登錄到用戶服務之后，Session中的用戶數據只會在用戶服務的應用中保存，而在其他服務中，并沒有對應的信息，但是我們現在希望的是，所有的服務都能夠同步這些Session信息，這樣我們才能實現在用戶服務登錄之后其他服務都能知道，那么我們該如何實現Session的同步呢？
那么，我們就著重來研究一下，然后實現2號方案，這里我們就使用Redis作為Session統一存儲，我們把一開始的壓縮包重新解壓一次，又來從頭開始編寫吧。
這里我們就只使用Nacos就行了，和之前一樣，我們把Nacos的包導入一下，然后進行一些配置：
現在我們需要為每個服務都添加驗證機制，首先導入依賴：
<!-- SpringSession Redis支持 --> <dependency><groupId>org.springframework.session</groupId><artifactId>spring-session-data-redis</artifactId> </dependency> <!-- 添加Redis的Starter --> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId> </dependency>然后我們依然使用SpringSecurity框架作為權限校驗框架：
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-security</artifactId> </dependency>接著我們在每個服務都編寫一下對應的配置文件：
spring:session:# 存儲類型修改為redisstore-type: redisredis:# Redis服務器的信息，該咋寫咋寫host: 1.14.121.107這樣，默認情況下，每個服務的接口都會被SpringSecurity所保護，只有登錄成功之后，才可以被訪問。
我們來打開Nacos看看：
可以看到三個服務都正常注冊了，接著我們去訪問圖書服務：
可以看到，訪問失敗，直接把我們給重定向到登陸頁面了，也就是說必須登陸之后才能訪問，同樣的方式去訪問其他服務，也是一樣的效果。
由于現在是統一Session存儲，那么我們就可以在任意一個服務登錄之后，其他服務都可以正常訪問，現在我們在當前頁面登錄，登錄之后可以看到圖書服務能夠正常訪問了：
同時用戶服務也能正常訪問了：
我們可以查看一下Redis服務器中是不是存儲了我們的Session信息：
雖然看起來好像確實沒啥問題了，但是借閱服務炸了，我們來看看為什么：
在RestTemplate進行遠程調用的時候，由于我們的請求沒有攜帶對應SESSION的Cookie，所以導致驗證失敗，訪問不成功，返回401，所以雖然這種方案看起來比較合理，但是在我們的實際使用中，還是存在一些不便的。
OAuth 2.0 實現單點登錄
**注意：**第一次接觸可能會比較難，不太好理解，需要多實踐和觀察。
前面我們雖然使用了統一存儲來解決Session共享問題，但是我們發現就算實現了Session共享，依然存在一些問題，由于我們每個服務都有自己的驗證模塊，實際上整個系統是存在冗余功能的、同時還有我們上面出現的問題，那么能否實現只在一個服務進行登錄，就可以訪問其他的服務呢？
實際上之前的登錄模式稱為多點登錄，而我們希望的是實現單點登陸，因此，我們得找一個更好的解決方案。
這里我們首先需要了解一種全新的登錄方式：OAuth 2.0，我們經?？吹揭恍┚W站支持第三方登錄，比如淘寶、咸魚我們就可以使用支付寶進行登錄，騰訊游戲可以用QQ或是微信登陸，以及微信小程序都可以直接使用微信進行登錄。我們知道它們并不是屬于同一個系統，比如淘寶和咸魚都不屬于支付寶這個應用，但是由于需要獲取支付寶的用戶信息，這時我們就需要使用 OAuth2.0 來實現第三方授權，基于第三方應用訪問用戶信息的權限（本質上就是給別人調用自己服務接口的權限），那么它是如何實現的呢？
四種授權模式
我們還是從理論開始講解，OAuth 2.0一共有四種授權模式：
客戶端模式（Client Credentials）
這是最簡單的一種模式，我們可以直接向驗證服務器請求一個Token（這里可能有些小伙伴對Token的概念不是很熟悉，Token相當于是一個令牌，我們需要在驗證服務器**（User Account And Authentication）**服務拿到令牌之后，才能去訪問資源，比如用戶信息、借閱信息等，這樣資源服務器才能知道我們是誰以及是否成功登錄了）
當然，這里的前端頁面只是一個例子，它還可以是其他任何類型的客戶端，比如App、小程序甚至是第三方應用的服務。
雖然這種模式比較簡便，但是已經失去了用戶驗證的意義，壓根就不是給用戶校驗準備的，而是更適用于服務內部調用的場景。
密碼模式（Resource Owner Password Credentials）
密碼模式相比客戶端模式，就多了用戶名和密碼的信息，用戶需要提供對應賬號的用戶名和密碼，才能獲取到Token。
雖然這樣看起來比較合理，但是會直接將賬號和密碼泄露給客戶端，需要后臺完全信任客戶端不會拿賬號密碼去干其他壞事，所以這也不是我們常見的。
隱式授權模式（Implicit Grant）
首先用戶訪問頁面時，會重定向到認證服務器，接著認證服務器給用戶一個認證頁面，等待用戶授權，用戶填寫信息完成授權后，認證服務器返回Token。
它適用于沒有服務端的第三方應用頁面，并且相比前面一種形式，驗證都是在驗證服務器進行的，敏感信息不會輕易泄露，但是Token依然存在泄露的風險。
授權碼模式（Authrization Code）
這種模式是最安全的一種模式，也是推薦使用的一種，比如我們手機上的很多App都是使用的這種模式。
相比隱式授權模式，它并不會直接返回Token，而是返回授權碼，真正的Token是通過應用服務器訪問驗證服務器獲得的。在一開始的時候，應用服務器（客戶端通過訪問自己的應用服務器來進而訪問其他服務）和驗證服務器之間會共享一個secret，這個東西沒有其他人知道，而驗證服務器在用戶驗證完成之后，會返回一個授權碼，應用服務器最后將授權碼和secret一起交給驗證服務器進行驗證，并且Token也是在服務端之間傳遞，不會直接給到客戶端。
這樣就算有人中途竊取了授權碼，也毫無意義，因為，Token的獲取必須同時攜帶授權碼和secret，但是secret第三方是無法得知的，并且Token不會直接丟給客戶端，大大減少了泄露的風險。
但是乍一看，OAuth 2.0不應該是那種第三方應用為了請求我們的服務而使用的嗎，而我們這里需要的只是實現同一個應用內部服務之間的認證，其實我也可以利用 OAuth2.0 來實現單點登錄，只是少了資源服務器這一角色，客戶端就是我們的整個系統，接下來就讓我們來實現一下。
搭建驗證服務器
第一步就是最重要的，我們需要搭建一個驗證服務器，它是我們進行權限校驗的核心，驗證服務器有很多的第三方實現也有Spring官方提供的實現，這里我們使用Spring官方提供的驗證服務器。
這里我們將最開始保存好的項目解壓，就重新創建一個新的項目，首先我們在父項目中添加最新的SpringCloud依賴：
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>2021.0.1</version><type>pom</type><scope>import</scope> </dependency>接著創建一個新的模塊auth-service，添加依賴：
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-security</artifactId></dependency><!-- OAuth2.0依賴，不再內置了，所以得我們自己指定一下版本 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-oauth2</artifactId><version>2.2.5.RELEASE</version></dependency> </dependencies>接著我們修改一下配置文件：
server:port: 8500servlet:#為了防止一會在服務之間跳轉導致Cookie打架（因為所有服務地址都是localhost，都會存JSESSIONID）#這里修改一下context-path，這樣保存的Cookie會使用指定的路徑，就不會和其他服務打架了#但是注意之后的請求都得在最前面加上這個路徑context-path: /sso接著我們需要編寫一下配置類，這里需要兩個配置類，一個是OAuth2的配置類，還有一個是SpringSecurity的配置類：
@Configuration public class SecurityConfiguration extends WebSecurityConfigurerAdapter {@Overrideprotected void configure(HttpSecurity http) throws Exception {http.authorizeRequests().anyRequest().authenticated() //.and().formLogin().permitAll(); //使用表單登錄}@Overrideprotected void configure(AuthenticationManagerBuilder auth) throws Exception {BCryptPasswordEncoder encoder = new BCryptPasswordEncoder();auth.inMemoryAuthentication() //直接創建一個用戶，懶得搞數據庫了.passwordEncoder(encoder).withUser("test").password(encoder.encode("123456")).roles("USER");}@Bean //這里需要將AuthenticationManager注冊為Bean，在OAuth配置中使用@Overridepublic AuthenticationManager authenticationManagerBean() throws Exception {return super.authenticationManagerBean();} } @EnableAuthorizationServer //開啟驗證服務器 @Configuration public class OAuth2Configuration extends AuthorizationServerConfigurerAdapter {@Resourceprivate AuthenticationManager manager;private final BCryptPasswordEncoder encoder = new BCryptPasswordEncoder();/*** 這個方法是對客戶端進行配置，一個驗證服務器可以預設很多個客戶端，* 之后這些指定的客戶端就可以按照下面指定的方式進行驗證* @param clients 客戶端配置工具*/@Overridepublic void configure(ClientDetailsServiceConfigurer clients) throws Exception {clients.inMemory() //這里我們直接硬編碼創建，當然也可以像Security那樣自定義或是使用JDBC從數據庫讀取.withClient("web") //客戶端名稱，隨便起就行.secret(encoder.encode("654321")) //只與客戶端分享的secret，隨便寫，但是注意要加密.autoApprove(false) //自動審批，這里關閉，要的就是一會體驗那種感覺.scopes("book", "user", "borrow") //授權范圍，這里我們使用全部all.authorizedGrantTypes("client_credentials", "password", "implicit", "authorization_code", "refresh_token");//授權模式，一共支持5種，除了之前我們介紹的四種之外，還有一個刷新Token的模式//這里我們直接把五種都寫上，方便一會實驗，當然各位也可以單獨只寫一種一個一個進行測試//現在我們指定的客戶端就支持這五種類型的授權方式了}@Overridepublic void configure(AuthorizationServerSecurityConfigurer security) {security.passwordEncoder(encoder) //編碼器設定為BCryptPasswordEncoder.allowFormAuthenticationForClients() //允許客戶端使用表單驗證，一會我們POST請求中會攜帶表單信息.checkTokenAccess("permitAll()"); //允許所有的Token查詢請求}@Overridepublic void configure(AuthorizationServerEndpointsConfigurer endpoints) {endpoints.authenticationManager(manager);//由于SpringSecurity新版本的一些底層改動，這里需要配置一下authenticationManager，才能正常使用password模式} }接著我們就可以啟動服務器了：
然后我們使用Postman進行接口測試，首先我們從最簡單的客戶端模式進行測試，客戶端模式只需要提供id和secret即可直接拿到Token，注意需要再添加一個grant_type來表明我們的授權方式，默認請求路徑為http://localhost:8500/sso/oauth/token：
發起請求后，可以看到我們得到了Token，它是以JSON格式給到我們的：
我們還可以訪問 http://localhost:8500/sso/oauth/check_token 來驗證我們的Token是否有效：
可以看到active為true，表示我們剛剛申請到的Token是有效的。
接著我們來測試一下第二種password模式，我們還需要提供具體的用戶名和密碼，授權模式定義為password即可：
接著我們需要在請求頭中添加Basic驗證信息，這里我們直接填寫id和secret即可：
可以看到在請求頭中自動生成了Basic驗證相關內容：
響應成功，得到Token信息，并且這里還多出了一個refresh_token，這是用于刷新Token的，我們之后會進行講解。
查詢Token信息之后還可以看到登錄的具體用戶以及角色權限等。
接著我們來看隱式授權模式，這種模式我們需要在驗證服務器上進行登錄操作，而不是直接請求Token，驗證登錄請求地址：http://localhost:8500/sso/oauth/authorize?client_id=web&response_type=token
注意response_type一定要是token類型，這樣才會直接返回Token，瀏覽器發起請求后，可以看到熟悉而又陌生的界面，沒錯，實際上這里就是使用我們之前講解的SpringSecurity進行登陸，當然也可以配置一下記住我之類的功能，這里就不演示了：
但是登錄之后我們發現出現了一個錯誤：
這是因為登錄成功之后，驗證服務器需要將結果給回客戶端，所以需要提供客戶端的回調地址，這樣瀏覽器就會被重定向到指定的回調地址并且請求中會攜帶Token信息，這里我們隨便配置一個回調地址：
@Override public void configure(ClientDetailsServiceConfigurer clients) throws Exception {clients.inMemory().withClient("web").secret(encoder.encode("654321")).autoApprove(false).scopes("book", "user", "borrow").redirectUris("http://localhost:8201/login") //可以寫多個，當有多個時需要在驗證請求中指定使用哪個地址進行回調.authorizedGrantTypes("client_credentials", "password", "implicit", "authorization_code", "refresh_token"); }接著重啟驗證服務器，再次訪問：
可以看到這里會讓我們選擇哪些范圍進行授權，就像我們在微信小程序中登陸一樣，會讓我們授予用戶信息權限、支付權限、信用查詢權限等，我們可以自由決定要不要給客戶端授予訪問這些資源的權限，這里我們全部選擇授予：
授予之后，可以看到瀏覽器被重定向到我們剛剛指定的回調地址中，并且攜帶了Token信息，現在我們來校驗一下看看：
可以看到，Token也是有效的。
最后我們來看看第四種最安全的授權碼模式，這種模式其實流程和上面是一樣的，但是請求的是code類型：http://localhost:8500/sso/oauth/authorize?client_id=web&response_type=code
可以看到訪問之后，依然會進入到回調地址，但是這時給的就是授權碼了，而不是直接給Token，那么這個Token該怎么獲取呢？
按照我們之前講解的原理，我們需要攜帶授權碼和secret一起請求，才能拿到Token，正常情況下是由回調的服務器進行處理，這里我們就在Postman中進行，我們復制剛剛得到的授權碼，接口依然是localhost:8500/sso/oauth/token：
可以看到結果也是正常返回了Token信息：
這樣我們四種最基本的Token請求方式就實現了。
最后還有一個是刷新令牌使用的，當我們的Token過期時，我們就可以使用這個refresh_token來申請一個新的Token：
但是執行之后我們發現會直接出現一個內部錯誤：
查看日志發現，這里還需要我們單獨配置一個UserDetailsService，我們直接把Security中的實例注冊為Bean：
@Bean @Override public UserDetailsService userDetailsServiceBean() throws Exception {return super.userDetailsServiceBean(); }然后在Endpoint中設置：
@Resource UserDetailsService service;@Override public void configure(AuthorizationServerEndpointsConfigurer endpoints) {endpoints.userDetailsService(service).authenticationManager(manager); }最后再次嘗試刷新Token：
OK，成功刷新Token，返回了一個新的。
基于@EnableOAuth2Sso實現
前面我們將驗證服務器已經搭建完成了，現在我們就來實現一下單點登陸吧，SpringCloud為我們提供了客戶端的直接實現，我們只需要添加一個注解和少量配置即可將我們的服務作為一個單點登陸應用，使用的是第四種授權碼模式。
一句話來說就是，這種模式只是將驗證方式由原本的默認登錄形式改變為了統一在授權服務器登陸的形式。
首先還是依賴：
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-security</artifactId> </dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-oauth2</artifactId><version>2.2.5.RELEASE</version> </dependency>我們只需要直接在啟動類上添加即可：
@EnableOAuth2Sso @SpringBootApplication public class BookApplication {public static void main(String[] args) {SpringApplication.run(BookApplication.class, args);} }我們不需要進行額外的配置類，因為這個注解已經幫我們做了：
@Target({ElementType.TYPE}) @Retention(RetentionPolicy.RUNTIME) @Documented @EnableOAuth2Client @EnableConfigurationProperties({OAuth2SsoProperties.class}) @Import({OAuth2SsoDefaultConfiguration.class, OAuth2SsoCustomConfiguration.class, ResourceServerTokenServicesConfiguration.class}) public @interface EnableOAuth2Sso { }可以看到它直接注冊了OAuth2SsoDefaultConfiguration，而這個類就是幫助我們對Security進行配置的：
@Configuration @Conditional({NeedsWebSecurityCondition.class}) public class OAuth2SsoDefaultConfiguration extends WebSecurityConfigurerAdapter {//直接繼承的WebSecurityConfigurerAdapter，幫我們把驗證設置都寫好了private final ApplicationContext applicationContext;public OAuth2SsoDefaultConfiguration(ApplicationContext applicationContext) {this.applicationContext = applicationContext;}接著我們需要在配置文件中配置我們的驗證服務器相關信息：
security:oauth2:client:#不多說了client-id: webclient-secret: 654321#Token獲取地址access-token-uri: http://localhost:8500/sso/oauth/token#驗證頁面地址user-authorization-uri: http://localhost:8500/sso/oauth/authorizeresource:#Token信息獲取和校驗地址token-info-uri: http://localhost:8500/sso/oauth/check_token現在我們就開啟圖書服務，調用圖書接口：
可以看到在發現沒有登錄驗證時，會直接跳轉到授權頁面，進行授權登錄，之后才可以繼續訪問圖書服務：
那么用戶信息呢？是否也一并保存過來了？我們這里直接獲取一下SpringSecurity的Context查看用戶信息，獲取方式跟我們之前的視頻中講解的是一樣的：
@RequestMapping("/book/{bid}") Book findBookById(@PathVariable("bid") int bid){//通過SecurityContextHolder將用戶信息取出SecurityContext context = SecurityContextHolder.getContext();System.out.println(context.getAuthentication());return service.getBookById(bid); }再次訪問圖書管理接口，可以看到：
這里使用的不是之前的UsernamePasswordAuthenticationToken也不是RememberMeAuthenticationToken，而是新的OAuth2Authentication，它保存了驗證服務器的一些信息，以及經過我們之前的登陸流程之后，驗證服務器發放給客戶端的Token信息，并通過Token信息在驗證服務器進行驗證獲取用戶信息，最后保存到Session中，表示用戶已驗證，所以本質上還是要依賴瀏覽器存Cookie的。
接下來我們將所有的服務都使用這種方式進行驗證，別忘了把重定向地址給所有服務都加上：
@Override public void configure(ClientDetailsServiceConfigurer clients) throws Exception {clients.inMemory().withClient("web").secret(encoder.encode("654321")).autoApprove(true) //這里把自動審批開了，就不用再去手動選同意了.scopes("book", "user", "borrow").redirectUris("http://localhost:8101/login", "http://localhost:8201/login", "http://localhost:8301/login").authorizedGrantTypes("client_credentials", "password", "implicit", "authorization_code", "refresh_token"); }這樣我們就可以實現只在驗證服務器登陸，如果登陸過其他的服務都可以訪問了。
但是我們發現一個問題，就是由于SESSION不同步，每次切換不同的服務進行訪問都會重新導驗證服務器去驗證一次：
這里有兩個方案：
- 像之前一樣做SESSION統一存儲
- 設置context-path路徑，每個服務單獨設置，就不會打架了
但是這樣依然沒法解決服務間調用的問題，所以僅僅依靠單點登陸的模式不太行。
基于@EnableResourceServer實現
前面我們講解了將我們的服務作為單點登陸應用直接實現單點登陸，那么現在我們如果是以第三方應用進行訪問呢？這時我們就需要將我們的服務作為資源服務了，作為資源服務就不會再提供驗證的過程，而是直接要求請求時攜帶Token，而驗證過程我們這里就繼續用Postman來完成，這才是我們常見的模式。
一句話來說，跟上面相比，我們只需要攜帶Token就能訪問這些資源服務器了，客戶端被獨立了出來，用于攜帶Token去訪問這些服務。
我們也只需要添加一個注解和少量配置即可：
@EnableResourceServer @SpringBootApplication public class BookApplication {public static void main(String[] args) {SpringApplication.run(BookApplication.class, args);} }配置中只需要：
security:oauth2:client:#基操client-id: webclient-secret: 654321resource:#因為資源服務器得驗證你的Token是否有訪問此資源的權限以及用戶信息，所以只需要一個驗證地址token-info-uri: http://localhost:8500/sso/oauth/check_token配置完成后，我們啟動服務器，直接訪問會發現：
這是由于我們的請求頭中沒有攜帶Token信息，現在有兩種方式可以訪問此資源：
- 在URL后面添加access_token請求參數，值為Token值
- 在請求頭中添加Authorization，值為Bearer +Token值
我們先來試試看最簡的一種：
另一種我們需要使用Postman來完成：
添加驗證信息后，會幫助我們轉換成請求頭信息：
這樣我們就將資源服務器搭建完成了。
我們接著來看如何對資源服務器進行深度自定義，我們可以為其編寫一個配置類，比如我們現在希望用戶授權了某個Scope才可以訪問此服務：
@Configuration public class ResourceConfiguration extends ResourceServerConfigurerAdapter { //繼承此類進行高度自定義@Overridepublic void configure(HttpSecurity http) throws Exception { //這里也有HttpSecurity對象，方便我們配置SpringSecurityhttp.authorizeRequests().anyRequest().access("#oauth2.hasScope('lbwnb')"); //添加自定義規則//Token必須要有我們自定義scope授權才可以訪問此資源} }可以看到當沒有對應的scope授權時，那么會直接返回insufficient_scope錯誤：
不知道各位是否有發現，實際上資源服務器完全沒有必要將Security的信息保存在Session中了，因為現在只需要將Token告訴資源服務器，那么資源服務器就可以聯系驗證服務器，得到用戶信息，就不需要使用之前的Session存儲機制了，所以你會發現HttpSession中沒有SPRING_SECURITY_CONTEXT，現在Security信息都是通過連接資源服務器獲取。
接著我們將所有的服務都
但是還有一個問題沒有解決，我們在使用RestTemplate進行服務間的遠程調用時，會得到以下錯誤：
實際上這是因為在服務調用時沒有攜帶Token信息，我們得想個辦法把用戶傳來的Token信息在進行遠程調用時也攜帶上，因此，我們可以直接使用OAuth2RestTemplate，它會在請求其他服務時攜帶當前請求的Token信息。它繼承自RestTemplate，這里我們直接定義一個Bean：
@Configuration public class WebConfiguration {@ResourceOAuth2ClientContext context;@Beanpublic OAuth2RestTemplate restTemplate(){return new OAuth2RestTemplate(new ClientCredentialsResourceDetails(), context);} }接著我們直接替換掉之前的RestTemplate即可：
@Service public class BorrowServiceImpl implements BorrowService {@ResourceBorrowMapper mapper;@ResourceOAuth2RestTemplate template;@Overridepublic UserBorrowDetail getUserBorrowDetailByUid(int uid) {List<Borrow> borrow = mapper.getBorrowsByUid(uid);User user = template.getForObject("http://localhost:8101/user/"+uid, User.class);//獲取每一本書的詳細信息List<Book> bookList = borrow.stream().map(b -> template.getForObject("http://localhost:8201/book/"+b.getBid(), Book.class)).collect(Collectors.toList());return new UserBorrowDetail(user, bookList);} }可以看到服務成功調用了：
現在我們來將Nacos加入，并通過Feign實現遠程調用。
依賴還是貼一下，不然找不到：
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-alibaba-dependencies</artifactId><version>2021.0.1.0</version><type>pom</type><scope>import</scope> </dependency> <dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-loadbalancer</artifactId> </dependency>所有服務都已經注冊成功了：
接著我們配置一下借閱服務的負載均衡：
@Configuration public class WebConfiguration {@ResourceOAuth2ClientContext context;@LoadBalanced //和RestTemplate一樣直接添加注解就行了@Beanpublic OAuth2RestTemplate restTemplate(){return new OAuth2RestTemplate(new ClientCredentialsResourceDetails(), context);} }現在我們來把它替換為Feign，老樣子，兩個客戶端：
@FeignClient("user-service") public interface UserClient {@RequestMapping("/user/{uid}")User getUserById(@PathVariable("uid") int uid); } @FeignClient("book-service") public interface BookClient {@RequestMapping("/book/{bid}")Book getBookById(@PathVariable("bid") int bid); }但是配置完成之后，又出現剛剛的問題了，OpenFeign也沒有攜帶Token進行訪問：
那么怎么配置Feign攜帶Token訪問呢？遇到這種問題直接去官方查：https://docs.spring.io/spring-cloud-openfeign/docs/current/reference/html/#oauth2-support，非常簡單，兩個配置就搞定：
feign:oauth2:#開啟Oauth支持，這樣就會在請求頭中攜帶Token了enabled: true#同時開啟負載均衡支持load-balanced: true重啟服務器，可以看到結果OK了：
這樣我們就成功將之前的三個服務作為資源服務器了，注意和我們上面的作為客戶端是不同的，將服務直接作為客戶端相當于只需要驗證通過即可，并且還是要保存Session信息，相當于只是將登錄流程換到統一的驗證服務器上進行罷了。而將其作為資源服務器，那么就需要另外找客戶端（可以是瀏覽器、小程序、App、第三方服務等）來訪問，并且也是需要先進行驗證然后再通過攜帶Token進行訪問，這種模式是我們比較常見的模式。
使用jwt存儲Token
官網：https://jwt.io
JSON Web Token令牌（JWT）是一個開放標準（RFC 7519），它定義了一種緊湊和自成一體的方式，用于在各方之間作為JSON對象安全地傳輸信息。這些信息可以被驗證和信任，因為它是數字簽名的。JWT可以使用密鑰（使用HMAC算法）或使用RSA或ECDSA進行公鑰/私鑰對進行簽名。
實際上，我們之前都是攜帶Token向資源服務器發起請求后，資源服務器由于不知道我們Token的用戶信息，所以需要向驗證服務器詢問此Token的認證信息，這樣才能得到Token代表的用戶信息，但是各位是否考慮過，如果每次用戶請求都去查詢用戶信息，那么在大量請求下，驗證服務器的壓力可能會非常的大。而使用JWT之后，Token中會直接保存用戶信息，這樣資源服務器就不再需要詢問驗證服務器，自行就可以完成解析，我們的目標是不聯系驗證服務器就能直接完成驗證。
JWT令牌的格式如下：
一個JWT令牌由3部分組成：標頭(Header)、有效載荷(Payload)和簽名(Signature)。在傳輸的時候，會將JWT的3部分分別進行Base64編碼后用.進行連接形成最終需要傳輸的字符串。
- 標頭：包含一些元數據信息，比如JWT簽名所使用的加密算法，還有類型，這里統一都是JWT。
- 有效載荷：包括用戶名稱、令牌發布時間、過期時間、JWT ID等，當然我們也可以自定義添加字段，我們的用戶信息一般都在這里存放。
- 簽名：首先需要指定一個密鑰，該密鑰僅僅保存在服務器中，保證不能讓其他用戶知道。然后使用Header中指定的算法對Header和Payload進行base64加密之后的結果通過密鑰計算哈希值，然后就得出一個簽名哈希。這個會用于之后驗證內容是否被篡改。
這里還是補充一下一些概念，因為很多東西都是我們之前沒有接觸過的：
-
**Base64：**就是包括小寫字母a-z、大寫字母A-Z、數字0-9、符號"+"、"/"一共64個字符的字符集（末尾還有1個或多個=用來湊夠字節數），任何的符號都可以轉換成這個字符集中的字符，這個轉換過程就叫做Base64編碼，編碼之后會生成只包含上述64個字符的字符串。相反，如果需要原本的內容，我們也可以進行Base64解碼，回到原有的樣子。
public void test(){String str = "你們可能不知道只用20萬贏到578萬是什么概念";//Base64不只是可以對字符串進行編碼，任何byte[]數據都可以，編碼結果可以是byte[]，也可以是字符串String encodeStr = Base64.getEncoder().encodeToString(str.getBytes());System.out.println("Base64編碼后的字符串："+encodeStr);System.out.println("解碼后的字符串："+new String(Base64.getDecoder().decode(encodeStr))); }注意Base64不是加密算法，只是一種信息的編碼方式而已。
-
**加密算法：加密算法分為對稱加密和非對稱加密，其中對稱加密（Symmetric Cryptography）**比較好理解，就像一把鎖配了兩把鑰匙一樣，這兩把鑰匙你和別人都有一把，然后你們直接傳遞數據，都會把數據用鎖給鎖上，就算傳遞的途中有人把數據竊取了，也沒辦法解密，因為鑰匙只有你和對方有，沒有鑰匙無法進行解密，但是這樣有個問題，既然解密的關鍵在于鑰匙本身，那么如果有人不僅竊取了數據，而且對方那邊的治安也不好，于是順手就偷走了鑰匙，那你們之間發的數據不就涼涼了嗎。
因此，**非對稱加密（Asymmetric Cryptography）**算法出現了，它并不是直接生成一把鑰匙，而是生成一個公鑰和一個私鑰，私鑰只能由你保管，而公鑰交給對方或是你要發送的任何人都行，現在你需要把數據傳給對方，那么就需要使用私鑰進行加密，但是，這個數據只能使用對應的公鑰進行解密，相反，如果對方需要給你發送數據，那么就需要用公鑰進行加密，而數據只能使用私鑰進行解密，這樣的話就算對方的公鑰被竊取，那么別人發給你的數據也沒辦法解密出來，因為需要私鑰才能解密，而只有你才有私鑰。
因此，非對稱加密的安全性會更高一些，包括HTTPS的隱私信息正是使用非對稱加密來保障傳輸數據的安全（當然HTTPS并不是單純地使用非對稱加密完成的，感興趣的可以去了解一下）
對稱加密和非對稱加密都有很多的算法，比如對稱加密，就有：DES、IDEA、RC2，非對稱加密有：RSA、DAS、ECC
-
不可逆加密算法：常見的不可逆加密算法有MD5, HMAC, SHA-1, SHA-224, SHA-256, SHA-384, 和SHA-512, 其中SHA-224、SHA-256、SHA-384，和SHA-512我們可以統稱為SHA2加密算法，SHA加密算法的安全性要比MD5更高，而SHA2加密算法比SHA1的要高，其中SHA后面的數字表示的是加密后的字符串長度，SHA1默認會產生一個160位的信息摘要。經過不可逆加密算法得到的加密結果，是無法解密回去的，也就是說加密出來是什么就是什么了。本質上，其就是一種哈希函數，用于對一段信息產生摘要，以防止被篡改。
實際上這種算法就常常被用作信息摘要計算，同樣的數據通過同樣的算法計算得到的結果肯定也一樣，而如果數據被修改，那么計算的結果肯定就不一樣了。
這里我們就可以利用jwt，將我們的Token采用新的方式進行存儲：
這里我們使用最簡單的一種方式，對稱密鑰，我們需要對驗證服務器進行一些修改：
@Bean public JwtAccessTokenConverter tokenConverter(){ //Token轉換器，將其轉換為JWTJwtAccessTokenConverter converter = new JwtAccessTokenConverter();converter.setSigningKey("lbwnb"); //這個是對稱密鑰，一會資源服務器那邊也要指定為這個return converter; }@Bean public TokenStore tokenStore(JwtAccessTokenConverter converter){ //Token存儲方式現在改為JWT存儲return new JwtTokenStore(converter); //傳入剛剛定義好的轉換器 } @Resource TokenStore store;@Resource JwtAccessTokenConverter converter;private AuthorizationServerTokenServices serverTokenServices(){ //這里對AuthorizationServerTokenServices進行一下配置DefaultTokenServices services = new DefaultTokenServices();services.setSupportRefreshToken(true); //允許Token刷新services.setTokenStore(store); //添加剛剛的TokenStoreservices.setTokenEnhancer(converter); //添加Token增強，其實就是JwtAccessTokenConverter，增強是添加一些自定義的數據到JWT中return services; }@Override public void configure(AuthorizationServerEndpointsConfigurer endpoints) {endpoints.tokenServices(serverTokenServices()) //設定為剛剛配置好的AuthorizationServerTokenServices.userDetailsService(service).authenticationManager(manager); }然后我們就可以重啟驗證服務器了：
可以看到成功獲取了AccessToken，但是這里的格式跟我們之前的格式就大不相同了，因為現在它是JWT令牌，我們可以對其進行一下Base64解碼：
可以看到所有的驗證信息包含在內，現在我們對資源服務器進行配置：
security:oauth2:resource:jwt:key-value: lbwnb #注意這里要跟驗證服務器的密鑰一致，這樣算出來的簽名才會一致然后啟動資源服務器，請求一下接口試試看：
請求成功，得到數據：
注意如果Token有誤，那么會得到：
Redis與分布式
在SpringBoot階段，我們學習了Redis，它是一個基于內存的高性能數據庫，我們當時已經學習了包括基本操作、常用數據類型、持久化、事務和鎖機制以及使用Java與Redis進行交互等，利用它的高性能，我們還使用它來做Mybatis的二級緩存、以及Token的持久化存儲。而這一部分，我們將繼續深入，探討Redis在分布式開發場景下的應用。
主從復制
在分布式場景下，我們可以考慮讓Redis實現主從模式：
主從復制，是指將一臺Redis服務器的數據，復制到其他的Redis服務器。前者稱為主節點(Master)，后者稱為從節點(Slave)，數據的復制是單向的，只能由主節點到從節點。Master以寫為主，Slave 以讀為主。
這樣的好處肯定是顯而易見的：
- 實現了讀寫分離，提高了性能。
- 在寫少讀多的場景下，我們甚至可以安排很多個從節點，這樣就能夠大幅度的分擔壓力，并且就算掛掉一個，其他的也能使用。
那么我們現在就來嘗試實現一下，這里我們還是在Windows下進行測試，打開Redis文件夾，我們要開啟兩個Redis服務器，修改配置文件redis.windows.conf：
# Accept connections on the specified port, default is 6379 (IANA #815344). # If port 0 is specified Redis will not listen on a TCP socket. port 6001一個服務器的端口設定為6001，復制一份，另一個的端口為6002，接著我們指定配置文件進行啟動，打開cmd：
現在我們的兩個服務器就啟動成功了，接著我們可以使用命令查看當前服務器的主從狀態，我們打開客戶端：
輸入info replication命令來查看當前的主從狀態，可以看到默認的角色為：master，也就是說所有的服務器在啟動之后都是主節點的狀態。那么現在我們希望讓6002作為從節點，通過一個命令即可：
可以看到，在輸入replicaof 127.0.0.1 6001命令后，就會將6001服務器作為主節點，而當前節點作為6001的從節點，并且角色也會變成：slave，接著我們來看看6001的情況：
可以看到從節點信息中已經出現了6002服務器，也就是說現在我們的6001和6002就形成了主從關系（還包含一個偏移量，這個偏移量反應的是從節點的同步情況）
主服務器和從服務器都會維護一個復制偏移量，主服務器每次向從服務器中傳遞 N 個字節的時候，會將自己的復制偏移量加上 N。從服務器中收到主服務器的 N 個字節的數據，就會將自己額復制偏移量加上 N，通過主從服務器的偏移量對比可以很清楚的知道主從服務器的數據是否處于一致，如果不一致就需要進行增量同步了。
那么我們現在可以來測試一下，在主節點新增數據，看看是否會同步到從節點：
可以看到，我們在6001服務器插入的a，可以在從節點6002讀取到，那么，從節點新增的數據在主節點能得到嗎？我們來測試一下：
可以看到，從節點壓根就沒辦法進行數據插入，節點的模式為只讀模式。那么如果我們現在不想讓6002作為6001的從節點了呢？
可以看到，通過輸入replicaof no one，即可變回Master角色。接著我們再來啟動一臺6003服務器，流程是一樣的：
可以看到，在連接之后，也會直接同步主節點的數據，因此無論是已經處于從節點狀態還是剛剛啟動完成的服務器，都會從主節點同步數據，實際上整個同步流程為：
當我們的主節點關閉后，從節點依然可以讀取數據：
但是從節點會瘋狂報錯：
當然每次都去敲個命令配置主從太麻煩了，我們可以直接在配置文件中配置，添加這樣行即可：
replicaof 127.0.0.1 6001這里我們給6002和6003服務器都配置一下，現在我們重啟三個服務器。
當然，除了作為Master節點的從節點外，我們還可以將其作為從節點的從節點，比如現在我們讓6003作為6002的從節點：
也就是說，現在差不多是這樣的的一個情況：
采用這種方式，優點肯定是顯而易見的，但是缺點也很明顯，整個傳播鏈路一旦中途出現問題，那么就會導致后面的從節點無法及時同步。
哨兵模式
前面我們講解了Redis實現主從復制的一些基本操作，那么我們接著來看哨兵模式。
經過之前的學習，我們發現，實際上最關鍵的還是主節點，因為一旦主節點出現問題，那么整個主從系統將無法寫入，因此，我們得想一個辦法，處理一下主節點故障的情況。實際上我們可以參考之前的服務治理模式，比如Nacos和Eureka，所有的服務都會被實時監控，那么只要出現問題，肯定是可以及時發現的，并且能夠采取響應的補救措施，這就是我們即將介紹的哨兵：
注意這里的哨兵不是我們之前學習SpringCloud Alibaba的那個，是專用于Redis的。哨兵會對所有的節點進行監控，如果發現主節點出現問題，那么會立即讓從節點進行投票，選舉一個新的主節點出來，這樣就不會由于主節點的故障導致整個系統不可寫（注意要實現這樣的功能最小的系統必須是一主一從，再小的話就沒有意義了）
那么怎么啟動一個哨兵呢？我們只需要稍微修改一下配置文件即可，這里直接刪除全部內容，添加：
sentinel monitor lbwnb 127.0.0.1 6001 1其中第一個和第二個是固定，第三個是為監控對象名稱，隨意，后面就是主節點的相關信息，包括IP地址和端口，最后一個1我們暫時先不說，然后我們使用此配置文件啟動服務器，可以看到啟動后：
可以看到以哨兵模式啟動后，會自動監控主節點，然后還會顯示那些節點是作為從節點存在的。
現在我們直接把主節點關閉，看看會發生什么事情：
可以看到從節點還是正常的在報錯，一開始的時候不會直接重新進行選舉而是繼續嘗試重連（因為有可能只是網絡小卡一下，沒必要這么敏感），但是我們發現，經過一段時間之后，依然無法連接，哨兵輸出了以下內容：
可以看到哨兵發現主節點已經有一段時間不可用了，那么就會開始進行重新選舉，6003節點被選為了新的主節點，并且之前的主節點6001變成了新的主節點的從節點：
當我們再次啟動6001時，會發現，它自動變成了6003的從節點，并且會將數據同步過來：
那么，這個選舉規則是怎樣的呢？是在所有的從節點中隨機選取還是遵循某種規則呢？
要是哨兵也掛了咋辦？沒事，咱們可以多安排幾個哨兵，只需要把哨兵的配置復制一下，然后修改端口，這樣就可以同時啟動多個哨兵了，我們啟動3個哨兵（一主二從三哨兵），這里我們吧最后一個值改為2：
sentinel monitor lbwnb 192.168.0.8 6001 2這個值實際上代表的是當有幾個哨兵認為主節點掛掉時，就判斷主節點真的掛掉了
現在我們把6001節點掛掉，看看這三個哨兵會怎么樣：
可以看到都顯示將master切換為6002節點了。
那么，在哨兵重新選舉新的主節點之后，我們Java中的Redis的客戶端怎么感知到呢？我們來看看，首先還是導入依賴：
<dependencies><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>4.2.1</version></dependency> </dependencies> public class Main {public static void main(String[] args) {//這里我們直接使用JedisSentinelPool來獲取Master節點//需要把三個哨兵的地址都填入try (JedisSentinelPool pool = new JedisSentinelPool("lbwnb",new HashSet<>(Arrays.asList("192.168.0.8:26741", "192.168.0.8:26740", "192.168.0.8:26739")))) {Jedis jedis = pool.getResource(); //直接詢問并得到Jedis對象，這就是連接的Master節點jedis.set("test", "114514"); //直接寫入即可，實際上就是向Master節點寫入Jedis jedis2 = pool.getResource(); //再次獲取System.out.println(jedis2.get("test")); //讀取操作} catch (Exception e) {e.printStackTrace();}} }這樣，Jedis對象就可以通過哨兵來獲取，當Master節點更新后，也能得到最新的。
集群搭建
如果我們服務器的內存不夠用了，但是現在我們的Redis又需要繼續存儲內容，那么這個時候就可以利用集群來實現擴容。
因為單機的內存容量最大就那么多，已經沒辦法再繼續擴展了，但是現在又需要存儲更多的內容，這時我們就可以讓N臺機器上的Redis來分別存儲各個部分的數據（每個Redis可以存儲1/N的數據量），這樣就實現了容量的橫向擴展。同時每臺Redis還可以配一個從節點，這樣就可以更好地保證數據的安全性。
那么問題來，現在用戶來了一個寫入的請求，數據該寫到哪個節點上呢？我們來研究一下集群的機制：
首先，一個Redis集群包含16384個插槽，集群中的每個Redis 實例負責維護一部分插槽以及插槽所映射的鍵值數據，那么這個插槽是什么意思呢？
實際上，插槽就是鍵的Hash計算后的一個結果，注意這里出現了計算機網絡中的CRC循環冗余校驗，這里采用CRC16，能得到16個bit位的數據，也就是說算出來之后結果是0-65535之間，再進行取模，得到最終結果：
Redis key的路由計算公式：slot = CRC16（key） % 16384
結果的值是多少，就應該存放到對應維護的Redis下，比如Redis節點1負責0-25565的插槽，而這時客戶端插入了一個新的數據a=10，a在Hash計算后結果為666，那么a就應該存放到1號Redis節點中。簡而言之，本質上就是通過哈希算法將插入的數據分攤到各個節點的，所以說哈希算法真的是處處都有用啊。
那么現在我們就來搭建一個簡單的Redis集群，這里創建6個配置，注意開啟集群模式：
# Normal Redis instances can't be part of a Redis Cluster; only nodes that are # started as cluster nodes can. In order to start a Redis instance as a # cluster node enable the cluster support uncommenting the following: # cluster-enabled yes接著記得把所有的持久化文件全部刪除，所有的節點內容必須是空的。
然后輸入redis-cli.exe --cluster create --cluster-replicas 1 127.0.0.1:6001 127.0.0.1:6002 127.0.0.1:6003 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003，這里的--cluster-replicas 1指的是每個節點配一個從節點：
輸入之后，會為你展示客戶端默認分配的方案，并且會詢問你當前的方案是否合理。可以看到6001/6002/6003都被選為主節點，其他的為從節點，我們直接輸入yes即可：
最后分配成功，可以看到插槽的分配情況：
現在我們隨便連接一個節點，嘗試插入一個值：
在插入時，出現了一個錯誤，實際上這就是因為a計算出來的哈希值（插槽），不歸當前節點管，我們得去管這個插槽的節點執行，通過上面的分配情況，我們可以得到15495屬于節點6003管理：
在6003節點插入成功，當然我們也可以使用集群方式連接，這樣我們無論在哪個節點都可以插入，只需要添加-c表示以集群模式訪問：
可以看到，在6001節點成功對a的值進行了更新，只不過還是被重定向到了6003節點進行插入。
我們可以輸入cluster nodes命令來查看當前所有節點的信息：
那么現在如果我們讓某一個主節點掛掉會怎么樣？現在我們把6001掛掉：
可以看到原本的6001從節點7001，晉升為了新的主節點，而之前的6001已經掛了，現在我們將6001重啟試試看：
可以看到6001變成了7001的從節點，那么要是6001和7001都掛了呢？
這時我們嘗試插入新的數據：
可以看到，當存在節點不可用時，會無法插入新的數據，現在我們將6001和7001恢復：
可以看到恢復之后又可以繼續正常使用了。
最后我們來看一下如何使用Java連接到集群模式下的Redis，我們需要用到JedisCluster對象：
public class Main {public static void main(String[] args) {//和客戶端一樣，隨便連一個就行，也可以多寫幾個，構造方法有很多種可以選擇try(JedisCluster cluster = new JedisCluster(new HostAndPort("192.168.0.8", 6003))){System.out.println("集群實例數量："+cluster.getClusterNodes().size());cluster.set("a", "yyds");System.out.println(cluster.get("a"));}} }操作基本和Jedis對象一樣，這里就不多做贅述了。
分布式鎖
在我們的傳統單體應用中，經常會用到鎖機制，目的是為了防止多線程競爭導致的并發問題，但是現在我們在分布式環境下，又該如何實現鎖機制呢？可能一條鏈路上有很多的應用，它們都是獨立運行的，這時我們就可以借助Redis來實現分布式鎖。
還記得我們上一章最后提出的問題嗎？
@Override public boolean doBorrow(int uid, int bid) {//1. 判斷圖書和用戶是否都支持借閱，如果此時來了10個線程，都進來了，那么都能夠判斷為可以借閱if(bookClient.bookRemain(bid) < 1)throw new RuntimeException("圖書數量不足");if(userClient.userRemain(uid) < 1)throw new RuntimeException("用戶借閱量不足");//2. 首先將圖書的數量-1，由于上面10個線程同時進來，同時判斷可以借閱，那么這個10個線程就同時將圖書數量-1，那庫存豈不是直接變成負數了？？？if(!bookClient.bookBorrow(bid))throw new RuntimeException("在借閱圖書時出現錯誤！");... }實際上在高并發下，我們看似正常的借閱流程，會出現問題，比如現在同時來了10個同學要借同一本書，但是現在只有3本，而我們的判斷規則是，首先看書夠不夠，如果此時這10個請求都已經走到這里，并且都判定為可以進行借閱，那么問題就出現了，接下來這10個請求都開始進行借閱操作，導致庫存直接爆表，形成超借問題（在電商系統中也存在同樣的超賣問題）
因此，為了解決這種問題，我們就可以利用分布式鎖來實現。那么Redis如何去實現分布式鎖呢？
在Redis存在這樣一個命令：
setnx key value這個命令看起來和set命令差不多，但是它有一個機制，就是只有當指定的key不存在的時候，才能進行插入，實際上就是set if not exists的縮寫。
可以看到，當客戶端1設定a之后，客戶端2使用setnx會直接失敗。
當客戶端1將a刪除之后，客戶端2就可以使用setnx成功插入了。
利用這種特性，我們就可以在不同的服務中實現分布式鎖，那么問題來了，要是某個服務加了鎖但是卡頓了呢，或是直接崩潰了，那這把鎖豈不是永遠無法釋放了？因此我們還可以考慮加個過期時間：
set a 666 EX 5 NX這里使用set命令，最后加一個NX表示是使用setnx的模式，和上面是一樣的，但是可以通過EX設定過期時間，這里設置為5秒，也就是說如果5秒還沒釋放，那么就自動刪除。
當然，添加了過期時間，帶了的好處是顯而易見的，但是同時也帶來了很多的麻煩，我們來設想一下這種情況：
因此，單純只是添加過期時間，會出現這種把別人加的鎖誰卸了的情況，要解決這種問題也很簡單，我們現在的目標就是保證任務只能刪除自己加的鎖，如果是別人加的鎖是沒有資格刪的，所以我們可以吧a的值指定為我們任務專屬的值，比如可以使用UUID之類的，如果在主動刪除鎖的時候發現值不是我們當前任務指定的，那么說明可能是因為超時，其他任務已經加鎖了。
如果你在學習本篇之前完成了JUC并發編程篇的學習，那么一定會有一個疑惑，如果在超時之前那一剎那進入到釋放鎖的階段，獲取到值肯定還是自己，但是在即將執行刪除之前，由于超時機制導致被刪除并且其他任務也加鎖了，那么這時再進行刪除，仍然會導致刪除其他任務加的鎖。
實際上本質還是因為鎖的超時時間不太好衡量，如果超時時間能夠設定地比較恰當，那么就可以避免這種問題了。
要解決這個問題，我們可以借助一下Redisson框架，它是Redis官方推薦的Java版的Redis客戶端。它提供的功能非常多，也非常強大，Redisson內部提供了一個監控鎖的看門狗，它的作用是在Redisson實例被關閉前，不斷的延長鎖的有效期，它為我們提供了很多種分布式鎖的實現，使用起來也類似我們在JUC中學習的鎖，這里我們嘗試使用一下它的分布式鎖功能。
<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.17.0</version> </dependency><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.75.Final</version> </dependency>首先我們來看看不加鎖的情況下：
public static void main(String[] args) {for (int i = 0; i < 10; i++) {new Thread(() -> {try(Jedis jedis = new Jedis("192.168.0.10", 6379)){for (int j = 0; j < 100; j++) { //每個客戶端獲取a然后增加a的值再寫回去，如果不加鎖那么肯定會出問題int a = Integer.parseInt(jedis.get("a")) + 1;jedis.set("a", a+"");}}}).start();} }這里沒有直接用incr而是我們自己進行計算，方便模擬，可以看到運行結束之后a的值并不是我們想要的：
現在我們來給它加一把鎖，注意這個鎖是基于Redis的，不僅僅只可以用于當前應用，是能夠垮系統的：
public static void main(String[] args) {Config config = new Config();config.useSingleServer().setAddress("redis://192.168.0.10:6379"); //配置連接的Redis服務器，也可以指定集群RedissonClient client = Redisson.create(config); //創建RedissonClient客戶端for (int i = 0; i < 10; i++) {new Thread(() -> {try(Jedis jedis = new Jedis("192.168.0.10", 6379)){RLock lock = client.getLock("testLock"); //指定鎖的名稱，拿到鎖對象for (int j = 0; j < 100; j++) {lock.lock(); //加鎖int a = Integer.parseInt(jedis.get("a")) + 1;jedis.set("a", a+"");lock.unlock(); //解鎖}}System.out.println("結束！");}).start();} }可以看到結果沒有問題：
注意，如果用于存放鎖的Redis服務器掛了，那么肯定是會出問題的，這個時候我們就可以使用RedLock，它的思路是，在多個Redis服務器上保存鎖，只需要超過半數的Redis服務器獲取到鎖，那么就真的獲取到鎖了，這樣就算掛掉一部分節點，也能保證正常運行，這里就不做演示了。
MySQL與分布式
前面我講解了Redis在分布式場景的下的相關應用，接著我們來看看MySQL數據庫在分布式場景下的應用。
主從復制
當我們使用MySQL的時候，也可以采取主從復制的策略，它的實現思路基本和Redis相似，也是采用增量復制的方式，MySQL會在運行的過程中，會記錄二進制日志，所有的DML和DDL操作都會被記錄進日志中，主庫只需要將記錄的操作復制給從庫，讓從庫也運行一次，那么就可以實現主從復制。但是注意它不會在一開始進行全量復制，所以最好再開始主從之前將數據庫的內容保持一致。
和之前一樣，一旦我們實現了主從復制，那么就算主庫出現故障，從庫也能正常提供服務，并且還可以實現讀寫分離等操作。這里我們就使用兩臺主機來搭建一主一從的環境，首先確保兩臺服務器都安裝了MySQL數據庫并且都已經正常運行了：
接著我們需要創建對應的賬號，一會方便從庫進行訪問的用戶：
CREATE USER test identified with mysql_native_password by '123456';接著我們開啟一下外網訪問：
sudo vim /etc/mysql/mysql.conf.d/mysqld.cnf修改配置文件：
# If MySQL is running as a replication slave, this should be # changed. Ref https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_tmpdir # tmpdir = /tmp # # Instead of skip-networking the default is now to listen only on # localhost which is more compatible and is not less secure. # bind-address = 127.0.0.1 這里注釋掉就行現在我們重啟一下MySQL服務：
sudo systemctl restart mysql.service現在我們首先來配置主庫，主庫只需要為我們剛剛創建好的用戶分配一個主從復制的權限即可：
grant replication slave on *.* to test; FLUSH PRIVILEGES;然后我們可以輸入命令來查看主庫的相關情況：
這樣主庫就搭建完成了，接著我們需要將從庫進行配置，首先是配置文件：
# The following can be used as easy to replay backup logs or for replication. # note: if you are setting up a replication slave, see README.Debian about # other settings you may need to change. # 這里需要將server-id配置為其他的值（默認是1）所有Mysql主從實例的id必須唯一，不能打架，不然一會開啟會失敗 server-id = 2進入數據庫，輸入：
change replication source to SOURCE_HOST='192.168.0.8',SOURCE_USER='test',SOURCE_PASSWORD='123456',SOURCE_LOG_FILE='binlog.000004',SOURCE_LOG_POS=591;注意后面的logfile和pos就是我們上面從主庫中顯示的信息。
執行完成后，顯示OK表示沒有問題，接著輸入：
start replica;現在我們的從機就正式啟動了，現在我們輸入：
show replica status\G;來查看當前從機狀態，可以看到：
最關鍵的是下面的Replica_IO_Running和Replica_SQL_Running必須同時為Yes才可以，實際上從庫會創建兩個線程，一個線程負責與主庫進行通信，獲取二進制日志，暫時存放到一個中間表（Relay_Log）中，而另一個線程則是將中間表保存的二進制日志的信息進行執行，然后插入到從庫中。
最后配置完成，我們來看看在主庫進行操作會不會同步到從庫：
可以看到在主庫中創建的數據庫，被同步到從庫中了，我們再來試試看創建表和插入數據：
use yyds; create table test (`id` int primary key,`name` varchar(255) NULL,`passwd` varchar(255) NULL );現在我們隨便插入一點數據：
這樣，我們的MySQL主從就搭建完成了，那么如果主機此時掛了會怎么樣？
可以看到IO線程是處于重連狀態，會等待主庫重新恢復運行。
分庫分表
在大型的互聯網系統中，可能單臺MySQL的存儲容量無法滿足業務的需求，這時候就需要進行擴容了。
和之前的問題一樣，單臺主機的硬件資源是存在瓶頸的，不可能無限制地縱向擴展，這時我們就得通過多臺實例來進行容量的橫向擴容，我們可以將數據分散存儲，讓多臺主機共同來保存數據。
那么問題來了，怎么個分散法？
-
**垂直拆分：**我們的表和數據庫都可以進行垂直拆分，所謂垂直拆分，就是將數據庫中所有的表，按照業務功能拆分到各個數據庫中（是不是感覺跟前面兩章的學習的架構對應起來了）而對于一張表，也可以通過外鍵之類的機制，將其拆分為多個表。
-
**水平拆分：**水平拆分針對的不是表，而是數據，我們可以讓很多個具有相同表的數據庫存放一部分數據，相當于是將數據分散存儲在各個節點上。
那么要實現這樣的拆分操作，我們自行去編寫代碼工作量肯定是比較大的，因此目前實際上已經有一些解決方案了，比如我們可以使用MyCat（也是一個數據庫中間件，相當于掛了一層代理，再通過MyCat進行分庫分表操作數據庫，只需要連接就能使用，類似的還有ShardingSphere-Proxy）或是Sharding JDBC（應用程序中直接對SQL語句進行分析，然后轉換成分庫分表操作，需要我們自己編寫一些邏輯代碼），這里我們就講解一下Sharding JDBC。
Sharding JDBC
**官方文檔（中文）：**https://shardingsphere.apache.org/document/5.1.0/cn/overview/#shardingsphere-jdbc
定位為輕量級 Java 框架，在 Java 的 JDBC 層提供的額外服務，它使用客戶端直連數據庫，以 jar 包形式提供服務，無需額外部署和依賴，可理解為增強版的 JDBC 驅動，完全兼容 JDBC 和各種 ORM 框架。
- 適用于任何基于 JDBC 的 ORM 框架，如：JPA, Hibernate, Mybatis, Spring JDBC Template 或直接使用 JDBC；
- 支持任何第三方的數據庫連接池，如：DBCP, C3P0, BoneCP, HikariCP 等；
- 支持任意實現 JDBC 規范的數據庫，目前支持 MySQL，PostgreSQL，Oracle，SQLServer 以及任何可使用 JDBC 訪問的數據庫。
這里我們主要演示一下水平分表方式，我們直接創建一個新的SpringBoot項目即可，依賴如下：
<dependencies><!-- ShardingJDBC依賴，那必須安排最新版啊，希望你們看視頻的時候還是5.X版本 --><dependency><groupId>org.apache.shardingsphere</groupId><artifactId>shardingsphere-jdbc-core-spring-boot-starter</artifactId><version>5.1.0</version></dependency><dependency><groupId>org.mybatis.spring.boot</groupId><artifactId>mybatis-spring-boot-starter</artifactId><version>2.2.2</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency> </dependencies>數據庫我們這里直接用上節課的即可，因為只需要兩個表結構一樣的數據庫即可，正好上節課進行了同步，所以我們直接把從庫變回正常狀態就可以了：
stop replica;接著我們把兩個表的root用戶密碼改一下，一會用這個用戶連接數據庫：
update user set authentication_string='' where user='root'; update user set host = '%' where user = 'root'; alter user root identified with mysql_native_password by '123456'; FLUSH PRIVILEGES;接著我們來看，如果直接嘗試開啟服務器，那肯定是開不了的，因為我們要配置數據源：
那么數據源該怎么配置呢？現在我們是一個分庫分表的狀態，需要配置兩個數據源：
spring:shardingsphere:datasource:# 有幾個數據就配幾個，這里是名稱，按照下面的格式，名稱+數字的形式names: db0,db1# 為每個數據源單獨進行配置db0:# 數據源實現類，這里使用默認的HikariDataSourcetype: com.zaxxer.hikari.HikariDataSource# 數據庫驅動driver-class-name: com.mysql.cj.jdbc.Driver# 不用我多說了吧jdbc-url: jdbc:mysql://192.168.0.8:3306/yydsusername: rootpassword: 123456db1:type: com.zaxxer.hikari.HikariDataSourcedriver-class-name: com.mysql.cj.jdbc.Driverjdbc-url: jdbc:mysql://192.168.0.13:3306/yydsusername: rootpassword: 123456如果啟動沒有問題，那么就是配置成功了：
接著我們需要對項目進行一些編寫，添加我們的用戶實體類和Mapper：
@Data @AllArgsConstructor public class User {int id;String name;String passwd; } @Mapper public interface UserMapper {@Select("select * from test where id = #{id}")User getUserById(int id);@Insert("insert into test(id, name, passwd) values(#{id}, #{name}, #{passwd})")int addUser(User user); }實際上這些操作都是常規操作，在編寫代碼時關注點依然放在業務本身上，現在我們就來編寫配置文件，我們需要告訴ShardingJDBC要如何進行分片，首先明確：現在是兩個數據庫都有test表存放用戶數據，我們目標是將用戶信息分別存放到這兩個數據庫的表中。
不廢話了，直接上配置：
spring:shardingsphere:rules:sharding:tables:#這里填寫表名稱，程序中對這張表的所有操作，都會采用下面的路由方案#比如我們上面Mybatis就是對test表進行操作，所以會走下面的路由方案test:#這里填寫實際的路由節點，比如現在我們要分兩個庫，那么就可以把兩個庫都寫上，以及對應的表#也可以使用表達式，比如下面的可以簡寫為 db$->{0..1}.testactual-data-nodes: db0.test,db1.test#這里是分庫策略配置database-strategy:#這里選擇標準策略，也可以配置復雜策略，基于多個鍵進行分片standard:#參與分片運算的字段，下面的算法會根據這里提供的字段進行運算sharding-column: id#這里填寫我們下面自定義的算法名稱sharding-algorithm-name: my-algsharding-algorithms:#自定義一個新的算法，名稱隨意my-alg:#算法類型，官方內置了很多種，這里演示最簡單的一種type: MODprops:sharding-count: 2props:#開啟日志，一會方便我們觀察sql-show: true其中，分片算法有很多內置的，可以在這里查詢：https://shardingsphere.apache.org/document/5.1.0/cn/user-manual/shardingsphere-jdbc/builtin-algorithm/sharding/，這里我們使用的是MOD，也就是取模分片算法，它會根據主鍵的值進行取模運算，比如我們這里填寫的是2，那么就表示對主鍵進行模2運算，根據數據源的名稱，比如db0就是取模后為0，db1就是取模后為1（官方文檔描述的并不是很清楚），也就是說，最終實現的效果就是單數放在db1，雙數放在db0，當然它還支持一些其他的算法，這里就不多介紹了。
那么現在我們編寫一個測試用例來看看，是否能夠按照我們上面的規則進行路由：
@SpringBootTest class ShardingJdbcTestApplicationTests {@ResourceUserMapper mapper;@Testvoid contextLoads() {for (int i = 0; i < 10; i++) {//這里ID自動生成1-10，然后插入數據庫mapper.addUser(new User(i, "xxx", "ccc")); }}}現在我們可以開始運行了：
測試通過，我們來看看數據庫里面是不是按照我們的規則進行數據插入的：
可以看到這兩張表，都成功按照我們指定的路由規則進行插入了，我們來看看詳細的路由情況，通過控制臺輸出的SQL就可以看到：
可以看到所有的SQL語句都有一個Logic SQL（這個就是我們在Mybatis里面寫的，是什么就是什么）緊接著下面就是Actual SQL，也就是說每個邏輯SQL最終會根據我們的策略轉換為實際SQL，比如第一條數據，它的id是0，那么實際轉換出來的SQL會在db0這個數據源進行插入。
這樣我們就很輕松地實現了分庫策略。
分庫完成之后，接著我們來看分表，比如現在我們的數據庫中有test_0和test_1兩張表，表結構一樣，但是我們也是希望能夠根據id取模運算的結果分別放到這兩個不同的表中，實現思路其實是差不多的，這里首先需要介紹一下兩種表概念：
- **邏輯表：**相同結構的水平拆分數據庫（表）的邏輯名稱，是 SQL 中表的邏輯標識。 例：訂單數據根據主鍵尾數拆分為 10 張表，分別是 t_order_0 到 t_order_9，他們的邏輯表名為 t_order
- **真實表：**在水平拆分的數據庫中真實存在的物理表。 即上個示例中的 t_order_0 到 t_order_9
現在我們就以一號數據庫為例，那么我們在里面創建上面提到的兩張表，之前的那個test表刪不刪都可以，就當做不存在就行了：
create table test_0 (`id` int primary key,`name` varchar(255) NULL,`passwd` varchar(255) NULL );create table test_1 (`id` int primary key,`name` varchar(255) NULL,`passwd` varchar(255) NULL );接著我們不要去修改任何的業務代碼，Mybatis里面寫的是什么依然保持原樣，即使我們的表名已經變了，我們需要做的是通過路由來修改原有的SQL，配置如下：
spring:shardingsphere:rules:sharding:tables:test:actual-data-nodes: db0.test_$->{0..1}#現在我們來配置一下分表策略，注意這里是table-strategy上面是database-strategytable-strategy:#基本都跟之前是一樣的standard:sharding-column: idsharding-algorithm-name: my-algsharding-algorithms:my-alg:#這里我們演示一下INLINE方式，我們可以自行編寫表達式來決定type: INLINEprops:#比如我們還是希望進行模2計算得到數據該去的表#只需要給一個最終的表名稱就行了test_，后面的數字是表達式取模算出的#實際上這樣寫和MOD模式一模一樣algorithm-expression: test_$->{id % 2}#沒錯，查詢也會根據分片策略來進行，但是如果我們使用的是范圍查詢，那么依然會進行全量查詢#這個我們后面緊接著會講，這里先寫上吧allow-range-query-with-inline-sharding: false現在我們來測試一下，看看會不會按照我們的策略進行分表插入：
可以看到，根據我們的算法，原本的邏輯表被修改為了最終進行分表計算后的結果，我們來查看一下數據庫：
插入我們了解完畢了，我們來看看查詢呢：
@SpringBootTest class ShardingJdbcTestApplicationTests {@ResourceUserMapper mapper;@Testvoid contextLoads() {System.out.println(mapper.getUserById(0));System.out.println(mapper.getUserById(1));}}可以看到，根據我們配置的策略，查詢也會自動選擇對應的表進行，是不是感覺有內味了。
那么如果是范圍查詢呢？
@Select("select * from test where id between #{start} and #{end}") List<User> getUsersByIdRange(int start, int end); @SpringBootTest class ShardingJdbcTestApplicationTests {@ResourceUserMapper mapper;@Testvoid contextLoads() {System.out.println(mapper.getUsersByIdRange(3, 5));}}我們來看看執行結果會怎么樣：
可以看到INLINE算法默認是不支持進行全量查詢的，我們得將上面的配置項改成true：
allow-range-query-with-inline-sharding: true再次進行測試：
可以看到，最終出來的SQL語句是直接對兩個表都進行查詢，然后求出一個并集出來作為最后的結果。
當然除了分片之外，還有廣播表和綁定表機制，用于多種業務場景下，這里就不多做介紹了，詳細請查閱官方文檔。
分布式序列算法
前面我們講解了如何進行分庫分表，接著我們來看看分布式序列算法。
在復雜分布式系統中，特別是微服構架中，往往需要對大量的數據和消息進行唯一標識。隨著系統的復雜，數據的增多，分庫分表成為了常見的方案，對數據分庫分表后需要有一個唯一ID來標識一條數據或消息（如訂單號、交易流水、事件編號等），此時一個能夠生成全局唯一ID的系統是非常必要的。
比如我們之前創建過學生信息表、圖書借閱表、圖書管理表，所有的信息都會有一個ID作為主鍵，并且這個ID有以下要求：
- 為了區別于其他的數據，這個ID必須是全局唯一的。
- 主鍵應該盡可能的保持有序，這樣會大大提升索引的查詢效率。
那么我們在分布式系統下，如何保證ID的生成滿足上面的需求呢？
**使用UUID：**UUID是由一組32位數的16進制數字隨機構成的，我們可以直接使用JDK為我們提供的UUID類來創建：
public static void main(String[] args) {String uuid = UUID.randomUUID().toString();System.out.println(uuid); }結果為73d5219b-dc0f-4282-ac6e-8df17bcd5860，生成速度非?？?#xff0c;可以看到確實是能夠保證唯一性，因為每次都不一樣，而且這么長一串那重復的概率真的是小的可憐。
但是它并不滿足我們上面的第二個要求，也就是說我們需要盡可能的保證有序，而這里我們得到的都是一些無序的ID。
雪花算法（Snowflake）：
我們來看雪花算法，它會生成一個一個64bit大小的整型的ID，int肯定是裝不下了。
可以看到它主要是三個部分組成，時間+工作機器ID+序列號，時間以毫秒為單位，41個bit位能表示約70年的時間，時間紀元從2016年11月1日零點開始，可以使用到2086年，工作機器ID其實就是節點ID，每個節點的ID都不相同，那么就可以區分出來，10個bit位可以表示最多1024個節點，最后12位就是每個節點下的序列號，因此每臺機器每毫秒就可以有4096個系列號。
這樣，它就兼具了上面所說的唯一性和有序性了，但是依然是有缺點的，第一個是時間問題，如果機器時間出現倒退，那么就會導致生成重復的ID，并且節點容量只有1024個，如果是超大規模集群，也是存在隱患的。
ShardingJDBC支持以上兩種算法為我們自動生成ID，文檔：https://shardingsphere.apache.org/document/5.1.0/cn/user-manual/shardingsphere-jdbc/builtin-algorithm/keygen/
這里，我們就是要ShardingJDBC來讓我們的主鍵ID以雪花算法進行生成，首先是配置數據庫，因為我們默認的id是int類型，裝不下64位的，改一下：
ALTER TABLE `yyds`.`test` MODIFY COLUMN `id` bigint NOT NULL FIRST;接著我們需要修改一下Mybatis的插入語句，因為現在id是由ShardingJDBC自動生成，我們就不需要自己加了：
@Insert("insert into test(name, passwd) values(#{name}, #{passwd})") int addUser(User user);接著我們在配置文件中將我們的算法寫上：
spring:shardingsphere:datasource:sharding:tables:test:actual-data-nodes: db0.test,db1.test#這里還是使用分庫策略database-strategy:standard:sharding-column: idsharding-algorithm-name: my-alg#這里使用自定義的主鍵生成策略key-generate-strategy:column: idkey-generator-name: my-genkey-generators:#這里寫我們自定義的主鍵生成算法my-gen:#使用雪花算法type: SNOWFLAKEprops:#工作機器ID，保證唯一就行worker-id: 666sharding-algorithms:my-alg:type: MODprops:sharding-count: 2接著我們來編寫一下測試用例：
@SpringBootTest class ShardingJdbcTestApplicationTests {@ResourceUserMapper mapper;@Testvoid contextLoads() {for (int i = 0; i < 20; i++) {mapper.addUser(new User("aaa", "bbb"));}}}可以看到日志：
在插入的時候，將我們的SQL語句自行添加了一個id字段，并且使用的是雪花算法生成的值，并且也是根據我們的分庫策略在進行插入操作。
讀寫分離
最后我們來看看讀寫分離，我們之前實現了MySQL的主從，那么我們就可以將主庫作為讀，從庫作為寫：
這里我們還是將數據庫變回主從狀態，直接刪除當前的表，我們重新來過：
drop table test;我們需要將從庫開啟只讀模式，在MySQL配置中進行修改：
read-only = 1這樣從庫就只能讀數據了（但是root賬號還是可以寫數據），接著我們重啟服務器：
sudo systemctl restart mysql.service然后進入主庫，看看狀態：
現在我們配置一下從庫：
change replication source to SOURCE_HOST='192.168.0.13',SOURCE_USER='test',SOURCE_PASSWORD='123456',SOURCE_LOG_FILE='binlog.000007',SOURCE_LOG_POS=19845; start replica;現在我們在主庫創建表：
create table test (`id` bigint primary key,`name` varchar(255) NULL,`passwd` varchar(255) NULL );然后我們就可以配置ShardingJDBC了，打開配置文件：
spring:shardingsphere:rules:#配置讀寫分離readwrite-splitting:data-sources:#名稱隨便寫user-db:#使用靜態類型，動態Dynamic類型可以自動發現auto-aware-data-source-name，這里不演示type: Staticprops:#配置寫庫（只能一個）write-data-source-name: db0#配置從庫（多個，逗號隔開）read-data-source-names: db1#負載均衡策略，可以自定義load-balancer-name: my-loadload-balancers:#自定義的負載均衡策略my-load:type: ROUND_ROBIN注意把之前改的用戶實體類和Mapper改回去，這里我們就不用自動生成ID的了。所有的負載均衡算法地址：https://shardingsphere.apache.org/document/5.1.0/cn/user-manual/shardingsphere-jdbc/builtin-algorithm/load-balance/
現在我們就來測試一下吧：
@SpringBootTest class ShardingJdbcTestApplicationTests {@ResourceUserMapper mapper;@Testvoid contextLoads() {mapper.addUser(new User(10, "aaa", "bbb"));System.out.println(mapper.getUserById(10));}}運行看看SQL日志：
可以看到，當我們執行插入操作時，會直接向db0進行操作，而讀取操作是會根據我們的配置，選擇db1進行操作。
至此，微服務應用章節到此結束。
總結
以上是生活随笔為你收集整理的SpringCloud笔记（三）微服务应用的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 直播代码
- 下一篇： 基于 Spring Boot + Clo