微服務初步學習

系統架構演變過程

一、單體架構

? ? ? ? 前后端都在一個項目中，包括我們現在的前后端分離開發，都可以看作是一個單體項目。

二、集群架構

? ? ? ? 把一個服務部署多次，可以解決服務不夠的問題，但是有些不必要的功能也跟著部署多次。

三、垂直架構

? ? ? ? 把不同的模塊進行拆分，可以單獨的部署訪問量大的模塊，模塊之間、服務之間沒有統一的管理，各自都是獨立的。

四、微服務架構

? ? ? ? 是一套完整的服務管理架構。

微服務的優勢

1、獨立開發：把服務進行拆分，不同的服務單獨開發。

2、獨立部署

3、故障隔離：一個服務可以部署多份，一臺服務器癱瘓，不會影響全局。

4、混合技術堆棧

5、粒度縮放：單個組件可以根據需要進行縮放，無需將所有組件放在一起

微服務中需要考慮到的問題？

1、多個小服務，如何對他們進行管理？（服務治理）

2、多個小服務，服務之間如何通訊？（服務調用）

3、多個小服務，客戶端如何訪問？（服務網關）

4、多個小服務，一旦出現了問題，應該如何自處理？

5、多個小服務，一旦出現了問題，應該如何排查錯誤？（鏈路追蹤）

微服務中常見的概念

服務治理

服務調用

服務網關

服務容錯

鏈路追蹤

MQ消息隊列

分布式鎖

分布式事務

服務治理

? ? ? ? 服務治理是服務注冊中心，所有的服務啟動后，都在注冊中心進行注冊，每一個服務都有一個自己的服務名稱。

常見的注冊中心：

Apache的Zookeeper、Spring Cloud的Eureka、Alibaba的Nacos

搭建Nacos環境

第一步：安裝nacos，下載地址：https://github.com/alibaba/nacos/releases，安裝nacos下載zip的格式的安裝包，然后進行解壓縮操作。

第二步：啟動nacos

? ? ? ? 切換目錄：cd nacos/bin

? ? ? ? 命令啟動：startup.cmd -m standalone

第三步：訪問nacos

打開瀏覽器輸入 http://localhost:8848/nacos,即可訪問服務，默認密碼是 nacos/nacos

將用戶微服務注冊到nacos

接下來將其注冊到nacos服務

1、在pom.xml中添加nacos依賴

<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency>

2、在啟動類上添加@EnableDiscoveryClient注解

3、在每個application.yml中為每一個微服務定義服務名，并添加nacos地址

spring:application:name: service-user #服務名cloud:nacos:discovery:server-addr: 127.0.0.1:8848 #nacos 地址

?4、啟動服務，觀察nacos的控制面板上是否有注冊上來的用戶微服務

同樣方法，將其他服務注冊到nacos

服務調用

使用nacos客戶端根據服務名動態獲取服務地址和端口

@AutowiredDiscoveryClient discoveryClient;

?從nacos中獲取服務地址

ServiceInstance serviceInstance =discoveryClient.getInstances("service-user").get(0);//getInstances獲取對應服務名的服務返回的是List<ServiceInstance>集合，獲取第0個服務
String purl = serviceInstance.getHost() + ":" +serviceInstance.getPort();
//使用
Product p = restTemplate.getForObject( "http://" + purl + "/product/get/"
+ pid, Product.class);

?服務調用負載均衡

? 什么是負載均衡？

? ? ? ? 就是將負載（工作任務，訪問請求）進行分攤到多個操作單元（服務器，組件）上進行執行。

自定義實現負載均衡

通過修改端口啟動兩個商品服務

server:port: 8091

server:port: 8092

?可以將獲取服務的方式改為隨機獲取

//獲取服務列表
List<ServiceInstance> instances =discoveryClient.getInstances("service-product");
//隨機生成索引
Integer index = new Random().nextInt(instances.size());
//獲取服務
ServiceInstance productService = instances.get(index);
//獲取服務地址
String purl = productService.getHost() + ":" +productService.getPort();

基于Ribbon實現負載均衡

Ribbon是Spring Cloud的一個組件，它可以讓我們使用一個注解就能輕松的搞定負載均衡。

在商品服務的application.yml中

ribbon:ConnectTimeout: 2000 # 請求連接的超時時間ReadTimeout: 5000 # 請求處理的超時時間
service-product: # 調用的提供者的名稱ribbon:NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule

第一步：在RestTemplate的生產方法上添加一個@LoadBlanced注解

第二步：修改服務調用的方法

restTemplate.getForObject("http://服務名/product/get/"+pid, Product.class);

七種負載均衡策略

1、輪詢策略：RoundRobinRule，按照一定的順序依次調用服務實例。比如一共3個服務，第一次調用服務1，第二次調用服務2，第三次調用服務3，依次類推。

NFLoadBalancerRuleClassName : com.netflix.loadbalancer.RoundRobinRule

2、權重策略：WeightedResponseTimeRule，根據每個服務提供者的響應時間分配一個權重，響應時間越長，權重越小，被選中的可能性也就越低。

實現原理是，剛開始使用輪詢策略并開啟一個計時器，每一段時間收集一次所有服務提供者的平均響應時間，然后再給每個服務提供者附上一個權重，權重越高被選中的概率越大。

NFLoadBalancerRuleClassName : com.netflix.loadbalancer.WeightedResponseTimeRule

?3、隨機策略：RandomRule，從服務提供者的列表中隨機安排一個服務實例。

NFLoadBalancerRuleClassName : com.netflix.loadbalancer.RandomRule

?4、最小連接數策略:BestAvailableRule，也叫最小并發數策略，它是遍歷服務者提供列表，選取連接數最小的服務實例。如果有相同的最小連接數，那么會調用輪詢策略進行選取。

NFLoadBalancerRuleClassName : com.netflix.loadbalancer.BestAvailableRule

?5、可用敏感性策略：AvailabilityFilteringRule，先過濾掉非健康的服務實例，然后再選擇連接數較小的服務實例。

NFLoadBalancerRuleClassName: com.netflix.loadbalancer.AvailabilityFilteringRule

?6、區域敏感策略：ZoneAvoidanceRule，根據服務所在區域(Zone)的性能和服務的可用性來選擇服務實例，在沒有區域的環境下，該策略和輪詢策略略類似。

7、重試策略：

Ribbon 的重試機制允許在服務請求失敗時，根據預先設定的策略進行重試。重試機制能夠顯著提高系統的容錯能力，尤其在網絡波動和服務臨時不可用的情況下，可以確保請求能夠成功完成。

Ribbon 重試機制主要包括兩種重試場景：

同一服務實例的重試：當請求某個服務實例時，如果請求失敗，Ribbon 會在同一個服務實例上進行多次重試。
切換服務實例的重試：如果在同一服務實例上重試多次后仍然失敗，Ribbon 會嘗試切換到其他可用的服務實例進行重試

ribbon:ConnectTimeout: 2000 # 請求連接的超時時間ReadTimeout: 5000 # 請求處理的超時時間
service-product: # 調用的提供者的名稱ribbon:MaxAutoRetries: 2  # 在同一個實例上的最大重試次數MaxAutoRetriesNextServer: 1  # 切換到下一個實例的最大重試次數OkToRetryOnAllOperations: true  # 對所有操作（包括非GET）進行重試ReadTimeout: 2000  # 讀超時ConnectTimeout: 1000  # 連接超時

基于Fegin實現服務調用

什么Fegin？

? ? ? ? Fegin是Spring Cloud提供的一個聲明式的偽Http客戶端，它使得調用遠程服務就像調用本地服務一樣簡單，只需要創建一個接口并添加一個注解即可。

? ? ? ? Nacos很好的兼容了Fegin，Fegin默認集成了Ribbon，所以在Nacos下使用Fegin默認實現了負載均衡的效果。

Fegin的使用

1、在使用Fegin的服務中導入Fegin依賴

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

2、在啟動類上添加Fegin的注解

3、創建對應服務的接口，并使用Fegin實現微服務調用

4、修改controller代碼，并啟動驗證

服務容錯

高并發情況下，如果訪問量過大，不加以控制，大量的請求堆積，會擊跨整個服務。

需要在某些場景下，為了保證服務不宕機，使用jmeter測試工具，模擬多線程，向后端發起請求。

要對請求進行限制? ?限流

Sentinel

Sentinel的主要功能就是容錯，主要體現在三個方面：

流量控制：限制每秒查詢的訪問量

熔斷降級：當檢測到調用鏈路中某個資源出現不穩定的表現，例如：請求響應時間長或異常比例升高的時候，則對這個資源的調用進行限制，讓請求快速失敗，避免影響到其他的資源而導致級聯故障。

Sentinel對這個問題采用兩種手段：

1、通過并發線程數進行限制

????????Sentinel 通過限制資源并發線程的數量，來減少不穩定資源對其它資源的影響。當某個資源出現不穩定的情況下，例如響應時間變長，對資源的直接影響就是會造成線程數的逐步堆積。當線程數在特定資源上堆積到一定的數量之后，對該資源的新請求就會被拒絕。堆積的線程完成任務后才開始繼續接收請求。

2、響應時間對資源進行降級

????????除了對并發線程數進行控制以外，Sentinel 還可以通過響應時間來快速降級不穩定的資源。當依賴的資源出現響應時間過長后，所有對該資源的訪問都會被直接拒絕，直到過了指定的時間窗口之后才重新恢復。

系統負載保護：

????????Sentinel 同時提供系統維度的自適應保護能力。當系統負載較高的時候，如果還持續讓請求進入可能會導致系統崩潰，無法響應。在集群環境下，會把本應這臺機器承載的流量轉發到其它的機器上去。如果這個時候其它的機器也處在一個邊緣狀態的時候，Sentinel 提供了對應的保護機制，讓系統的入口流量和系統的負載達到一個平衡，保證系統在能力范圍之內處理最多的請求。

總之一句話: 我們需要做的事情，就是在 Sentinel 的資源上配置各種各樣的規 則，來實現各種容錯的功能。

實現一個接口的限流

為某個接口添加訪問控制

訪問數量超時限流

?服務網關

在微服務架構中有很多服務，那么客戶端要記錄很多的服務地址，在每一個服務中都要進行權限認證、授權等操作。

解決方案：

在微服務架構中，我們提供一個 統一訪問入口---網關

在網關中做倆件事情：

1、全局過濾器

2、在網關中進行限流

后面訪問的地址：網關地址+服務名+接口地址

消息隊列MQ

什么是MQ？

? ? ? ? Message Queue：消息隊列，是一個組件，是一種提供消息隊列服務的中間件，也稱消息中間件，是一套提供了消息生產、存儲、消費全過程API的軟件系統（消息即數據），簡單來說就是一個先進先出的數據結構。

常見的MQ應用場景：

異步解耦：

? ? ? ? 常見的一個場景是用戶注冊后，需要發送郵件和短信通知，已告知用戶注冊成功。傳統的做法如下：

? ? ? ? 此架構下注冊、郵件、短信三個任務全部完成后，才會注冊結構到客戶端，用戶才可以使用賬號登錄。但是對與用戶來說，注冊功能實際只需要注冊系統存儲用戶的信息后，用戶就可以登錄，而后續的注冊短信和郵件不是及時需要關注的，所以實際當數據寫入注冊系統后，注冊系統就可以把其他的操作放入對應的消息隊列MQ中然后返回用戶結果，有消息隊列MQ異步進行其他操作。

? ? ? ? 異步解耦是消息隊列MQ的主要特點，主要目的是減少請求響應時間和解耦。主要的使用場景就是將比叫消耗時間且不需要及時同步返回結果的操作放入消息隊列。同時，由于使用了消息隊列MQ，只要保證消息格式不變，消息的發送方和接收方并不需要彼此聯系，也不需要接收對方的影響，即解耦。

RocketMQ架構

NameServer：消息隊列的協調者，Broker向它注冊路由信息，同時Producer和Consumer向其獲取路由信息

Broker：是RocketMQ的核心，負責消息的接收，存儲，發送等功能

Producer：消息的生產者，需要從NameServer獲取Boker的信息，然后與Broker建立連接，向Broker發送消息

Consumer：消息的接收者，需要從NameServer獲取Broker信息，然后與Broker建立連接，從Broker中獲取消息。

Topic：用來區分不同類型的消息，發送和接收消息前都需要先創建Topic，針對Topic來發送和接收消息

Message Queue：為了提高性能和吞吐量，引入了Message Queue，一個Topic可以設置一個或多個Message Queue，這樣消息就可以并行往各個Message Queue發送消息，消費者也可以并行的從多個Message Queue讀取消息。就是消息的載體。

Producer Group：生產者組，簡單來說就是多個發送同一類消息的生產者。

Consumer Group：接收者組，接收同一類消息的多個consumer實例組成的。

java代碼演示消息發送和接收

導入依賴

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.2</version>
</dependency>

發送消息：

1、創建消息生產者，指定生產者所屬的組名

2、指定NameServer地址

3、啟動生產者

4、創建消息對象，指定主題、標簽、消息體。

5、發送消息

6、關閉生產者

package org.example;import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;public class Productor {public static void main(String[] args) throws Exception {
//1. 創建消息生產者, 指定生產者所屬的組名DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
//2. 指定 Nameserver 地址producer.setNamesrvAddr("127.0.0.1:9876");
//3. 啟動生產者producer.start();
//4. 創建消息對象，指定主題、標簽和消息體Message msg = new Message("myTopic", "myTag",("RocketMQ Message哇哇哇哇").getBytes());Message msg2 = new Message("myTopic", "myTa",("RocketMQ Message哇哇哇哇222").getBytes());
//5. 發送消息SendResult sendResult = producer.send(msg, 10000);SendResult send = producer.send(msg2, 10000);System.out.println(sendResult);System.out.println(send);
//6. 關閉生產者producer.shutdown();}
}

接收消息

1、創建消息接收者，指定消費者所屬的組名

2、指定NameServer地址

3、指定接收者訂閱的主題和標簽

4、設置回調函數，編寫處理消息的方法

5、啟動消息接收者

package org.example;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class Comsumer {public static void main(String[] args) throws Exception {
//1. 創建消息消費者, 指定消費者所屬的組名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumergroup");
//2. 指定 Nameserver 地址consumer.setNamesrvAddr("127.0.0.1:9876");
//3. 指定消費者訂閱的主題和標簽consumer.subscribe("myTopic", "*");
//4. 設置回調函數，編寫處理消息的方法consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.println("Receive New Messages: " + new String(msgs.get(0).getBody()));//返回消費狀態return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});
//5. 啟動消息消費者consumer.start();System.out.println("Consumer Started.");}
}

在微服務中RocketMQ的使用

訂單服務是發送消息

添加依賴

<!--rocketmq-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.2</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>

添加配置

rocketmq:
name-server: 127.0.0.1:9876 #rocketMQ 服務的地址
producer:
group: shop-order # 生產者組

?編寫測試代碼

@Autowired
private RocketMQTemplate rocketMQTemplate;//主題，消息
rocketMQTemplate.convertAndSend("order-topic", order);

用戶服務接收消息

添加依賴

<!--rocketMQ-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.2</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>

修改配置文件

rocketmq:
name-server: 127.0.0.1:9876

編寫消息接收服務

@Service
@RocketMQMessageListener(consumerGroup = "shop-user", topic = "order-topic")
public class SmsService implements RocketMQListener<Order> {
@Override
public void onMessage(Order order) {
System.out.println("收到一個訂單信息:"+ JSON.toJSONString(order)+",接
下來發送短信");
}
}

啟動服務，執行下單操作，查看控制臺

成功接收到消息。

Redis實現分布式鎖

什么是分布式鎖？

? ? ? ? 即分布式系統中的鎖，在單體項目中我們通過Java中的鎖解決多線程訪問共享資源的問題，而分布式鎖，解決了分布式系統中控制共享資源訪問的問題。與單體項目不同的是，分布式系統中競爭共享資源的最小粒度從線程升級到了進程。

為什么需要分布式鎖？

? ? ? ? 在分布式微服務架構中，一個應用往往需要開啟多個服務（每一個服務都是一個獨立的進程），這樣依賴Java中的鎖synchronized和Lock失效了，只有在同一個進程是有效的。

如何實現分布式鎖

可以通過redis實現分布式鎖，在redis中存放一個標志，當一個請求到達時修改標志

方式一：redis+setnx命令，自己實現一個分布式鎖，雖然可以實現，在一些簡單的場景下，沒有問題的，在復雜的情況下還是會出現問題的。

setnx key value 設置鍵值時，會先判斷鍵是否存在，鍵如果不存在，設置成功，鍵如果存在，就設置失敗

如果不將鎖釋放當代finally中，就會造成死鎖，原因是1、程序處理邏輯出現了問題，沒有及時釋放鎖2、進程掛了，沒有機會釋放鎖。

@RequestMapping("/sub")public void sub(){// 使用redis，以及redis中的setnx命令實現分布式鎖// setnx key value 設置鍵值時，會先判斷鍵是否存在，鍵如果不存在，設置成功，鍵如果存在，就設置失敗//設置失效時間，第一個線程進來后就會，就會獲取鎖，第一個線程執行的業務超過失效時間，key就會失效，其他線程進來獲取鎖//第一個線程業務執行完之后就回去釋放鎖，那么鎖就會釋放成其他線程的鎖，所以設置失效時間也是會出現問題的。try{// 針對進程可能掛掉的情況，為ket設置一個失效時間，即使服務掛了，也會刪除keyboolean res = redisTemplate.opsForValue().setIfAbsent("lock",10, TimeUnit.SECONDS);if (!res){return;}// 從數據庫查詢一下庫存Integer stock = (Integer) redisTemplate.opsForValue().get("");// 如果庫存>0，就扣庫存if (stock>0){Integer real = stock-1;redisTemplate.opsForValue().set("stock", real); // 把釋放鎖寫在finally中，即使出現異常也，會釋放鎖System.out.println("成功");}else {System.out.println("扣庫存失敗");}}finally {redisTemplate.delete("lock");}}

在finally中釋放鎖以及設置鍵的失效時間，也是會出現問題的。

例如：程序的業務邏輯執行時間大于失效時間，那么鎖就會失效，其他線程就會進來，這時，當第一個線程執行完成之后，會誤刪除第二個線程的鎖的，導致其他線程的鎖失效。

解決辦法：為每一個線程添加一個版本號，刪除鎖時判斷版本號。

 @RequestMapping("/sub")public void sub(){// 使用redis，以及redis中的setnx命令實現分布式鎖// setnx key value 設置鍵值時，會先判斷鍵是否存在，鍵如果不存在，設置成功，鍵如果存在，就設置失敗//設置失效時間，第一個線程進來后就會，就會獲取鎖，第一個線程執行的業務超過失效時間，key就會失效，其他線程進來獲取鎖//第一個線程業務執行完之后就回去釋放鎖，那么鎖就會釋放成其他線程的鎖，所以設置失效時間也是會出現問題的。String createId = UUID.randomUUID().toString();//生產一個32位不重復的字符串，版本號try{// 針對進程可能掛掉的情況，為ket設置一個失效時間，即使服務掛了，也會刪除keyboolean res = redisTemplate.opsForValue().setIfAbsent("lock",createId ,10, TimeUnit.SECONDS);if (!res){return;}// 從數據庫查詢一下庫存Integer stock = (Integer) redisTemplate.opsForValue().get("");// 如果庫存>0，就扣庫存if (stock>0){Integer real = stock-1;redisTemplate.opsForValue().set("stock", real); // 把釋放鎖寫在finally中，即使出現異常也，會釋放鎖System.out.println("成功");}else {System.out.println("扣庫存失敗");}}finally {String lock=(String)redisTemplate.opsForValue().get("lock");if (createId.equals(lock)){//版本號相同，可以釋放鎖，不一樣則說明已經過期了，則不需要管了。redisTemplate.delete("lock");}}}

?方式2：使用redisson

導入依賴

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.6.5</version></dependency>

?創建redisson對象

package com.ffyc.springcloudshop.config;import org.redisson.Redisson;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedissonConfig {//創建 Redisson 對象@Beanpublic Redisson getRedisson(){Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);return (Redisson)Redisson.create(config);}}

使用redisson實現加鎖釋放鎖

 @RequestMapping("/sub")public void sub(){RLock lock=redisson.getLock("stock-lock");//定義redis中鎖的標志keytry {lock.lock(30,TimeUnit.SECONDS);//獲取鎖// 從數據庫查詢一下庫存Integer stock = (Integer) redisTemplate.opsForValue().get("stock");// 如果庫存>0，就扣庫存if (stock>0){Integer real = stock-1;redisTemplate.opsForValue().set("stock", real); // 把釋放鎖寫在finally中，即使出現異常也，會釋放鎖System.out.println("成功");}else {System.out.println("扣庫存失敗");}}finally {lock.unlock();//釋放鎖}}