一、背景

在微服務架構中 API網關 非常重要，網關作為全局流量入口並不單單是一個反向路由，更多的是把各個邊緣服務(Web層)的各種共性需求抽取出來放在一個公共的“服務”(網關)中實現，例如安全認證、權限控制、限流熔斷、監控、跨域處理、聚合API文檔等公共功能。

 

在以 Dubbo 框架體系來構建的微服務架構下想要增加API網關，如果不想自研開發的情況下在目前的開源社區中幾乎沒有找到支持dubbo協議的主流網關，但是 Spring Cloud 體系下卻有兩個非常熱門的開源API網關可以選擇；本文主要介紹如何通過 Nacos 整合 Spring Cloud Gateway 與 Dubbo 服務。

 

二、傳統 dubbo 架構

dubbo屬於rpc調用，所以必須提供一個web層的服務作為http入口給客戶端調用，並在上面提供安全認證等基礎功能，而web層前面對接Nginx等反向代理用於統一入口和負載均衡。

web層一般是根據業務模塊來切分的，用於聚合某個業務模塊所依賴的各個service服務

PS：我們能否把上圖中的web層全部整合在一起成為一個API網關呢？(不建議這樣做)

因為這樣的web層並沒有實現 泛化調用 必須引入所有dubbo服務的api依賴，會使得網關變得非常不穩定，任何服務的接口變更都需要修改網關中的api依賴！

 

三、整合 Srping Cloud Gateway 網關

下面就開始聊聊直接拿熱門的 Srping Cloud Gateway 來作為dubbo架構體系的網關是否可行，首先該API網關是屬於 Spring Cloud 體系下的組件之一，要整合dubbo的話需要解決以下問題：

1. 打通註冊中心：spring cloud gateway 需要通過註冊中心發現下游服務，而 dubbo 也需要通過註冊中心實現服務的註冊與發現，如果兩者的註冊中心不能打通的話就會變成雙註冊中心架構就非常複雜了！
2. 協議轉換： gateway 使用http傳輸協議調用下游服務，而dubbo服務默認使用的是tcp傳輸協議

上面提到的第一個問題“打通註冊中心”其實已經不是問題了，目前dubbo支持 Zookeeper 與 Nacos 兩個註冊中心，而 Spring Cloud 自從把 @EnableEurekaClient 改為 @EnableDiscoveryClient 之後已經基本上支持所有主流的註冊中心了，本文將使用 Nacos 作為註冊中心打通兩者

 

3.1. 方式一

把傳統dubbo架構中的 Nginx 替換為 Spring Cloud Gateway ，並把 安全認證 等共性功能前移至網關處實現

由於web層服務本身提供的就是http接口，所以網關層無需作協議轉換，但是由於 安全認證 前移至網關了需要通過網絡隔離的手段防止被繞過網關直接請求後面的web層

 

3.2. 方式二

dubbo服務本身修改或添加 rest 傳輸協議的支持，這樣網關就可以通過http傳輸協議與dubbo服務通信了

rest傳輸協議：基於標準的Java REST API——JAX-RS 2.0（Java API for RESTful Web Services的簡寫）實現的REST調用支持

目前版本的dubbo已經支持dubbo、rest、rmi、hessian、http、webservice、thrift、redis等10種傳輸協議了，並且還支持同一個服務同時定義多種協議，例如配置 protocol = { “dubbo”, “rest” } 則該服務同時支持 dubbo 與 rest 兩種傳輸協議

 

3.3. 總結

方式一 對比 方式二 多了一層web服務所以多了一次網絡調用開銷，但是優點是各自的職責明確單一，web層可以作為聚合層用於聚合多個service服務的結果經過融合加工一併返回給前端，所以這種架構下能大大減少服務的 循環依賴

 

四、代碼實踐

依賴環境

• lombok
• jdk 1.8
• Nacos 1.3
• Spring Boot 2.2.8.RELEASE
• Spring Cloud Hoxton.SR5
• Spring Cloud Alibaba 2.2.1.RELEASE

 

在根目錄的 pom.xml 中定義全局的依賴版本

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <java.version>8</java.version>

        <spring-boot-dependencies.version>2.2.8.RELEASE</spring-boot-dependencies.version>
        <spring-cloud-dependencies.version>Hoxton.SR5</spring-cloud-dependencies.version>
        <spring-cloud-alibaba-dependencies.version>2.2.1.RELEASE</spring-cloud-alibaba-dependencies.version>
        <jaxrs.version>3.12.1.Final</jaxrs.version>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot-dependencies.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud-dependencies.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>${spring-cloud-alibaba-dependencies.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

 

4.1. 創建dubbo-api工程

分別定義兩個api接口

DubboService 使用dubbo協議的服務

public interface DubboService {
    String test(String param);
}

RestService 使用rest協議的服務

public interface RestService {
    String test(String param);
}

 

4.2. 創建web-dubbo工程

使用 方式一 整合對接網關，這裏為了簡化在同一個服務下只使用邏輯分層定義controller層與service層，並沒有做服務拆分

4.2.1. 創建配置

定義 spring boot 配置

server:
  port: 8081

spring:
  application:
    name: zlt-web-dubbo
  main:
    allow-bean-definition-overriding: true
  cloud:
    nacos:
      server-addr: 192.168.28.130:8848
      username: nacos
      password: nacos

server.port：配置應用服務器暴露的端口

spring.cloud.nacos：配置 spring cloud 的註冊中心相關參數，nacos 的配置需要改為自己環境所對應

定義 dubbo 配置

dubbo:
  scan:
    base-packages: org.zlt.service
  protocols:
    dubbo:
      name: dubbo
      port: -1
  registry:
    address: spring-cloud://localhost
  consumer:
    timeout: 5000
    check: false
    retries: 0
  cloud:
    subscribed-services:

dubbo.scan.base-packages：指定 Dubbo 服務實現類的掃描基準包

dubbo.protocols：服務暴露的協議配置，其中子屬性 name 為協議名稱，port 為協議端口（ -1 表示自增端口，從 20880 開始）

dubbo.registry.address：Dubbo 服務註冊中心配置，其中子屬性 address 的值 “spring-cloud://localhost”，說明掛載到 Spring Cloud 註冊中心

 

4.2.2. 創建DubboService的實現類

通過 protocol = "dubbo" 指定使用 dubbo協議 定義服務

@Service(protocol = "dubbo")
public class DubboServiceImpl implements DubboService {
    @Override
    public String test(String param) {
        return "dubbo service: " + param;
    }
}

 

4.2.3. 創建Controller類

使用 Spring Boot 的 @RestController 註解定義web服務

@RestController
public class WebController {
    @Autowired
    private DubboService dubboService;

    @GetMapping("/test/{p}")
    public String test(@PathVariable("p") String param) {
        return dubboService.test(param);
    }
}

 

4.3. 創建rest-dubbo工程

使用 方式二 整合對接網關，由於該服務是通過dubbo來創建rest服務，所以並不需要使用 Spring Boot 內置應用服務

4.3.1. 創建配置

定義 spring boot 配置

spring:
  application:
    name: zlt-rest-dubbo
  main:
    allow-bean-definition-overriding: true
  cloud:
    nacos:
      server-addr: 192.168.28.130:8848
      username: nacos
      password: nacos

因為不使用 Spring Boot 內置的應用服務所以這裏並不需要指定 server.port

定義 dubbo 配置

dubbo:
  scan:
    base-packages: org.zlt.service
  protocols:
    dubbo:
      name: dubbo
      port: -1
    rest:
      name: rest
      port: 8080
      server: netty
  registry:
    address: spring-cloud://localhost
  consumer:
    timeout: 5000
    check: false
    retries: 0
  cloud:
    subscribed-services:

dubbo.protocols：配置兩種協議，其中rest協議定義 8080 端口並使用 netty 作為應用服務器

 

4.3.2. 創建RestService的實現類

通過 protocol = "rest" 指定使用 rest協議 定義服務

@Service(protocol = "rest")
@Path("/")
public class RestServiceImpl implements RestService {
    @Override
    @Path("test/{p}")
    @GET
    public String test(@PathParam("p") String param) {
        return "rest service: " + param;
    }
}

 

4.4. 創建Spring Cloud Gateway工程

定義 spring boot 配置

server:
  port: 9900

spring:
  application:
    name: sc-gateway
  main:
    allow-bean-definition-overriding: true
  cloud:
    nacos:
      server-addr: 192.168.28.130:8848
      username: nacos
      password: nacos

server.port：定義網關端口為 9090

定義網關配置

spring:
  cloud:
    gateway:
      discovery:
        locator:
          lowerCaseServiceId: true
          enabled: true
      routes:
        - id: web
          uri: lb://zlt-web-dubbo
          predicates:
            - Path=/api-web/**
          filters:
            - StripPrefix=1
        - id: rest
          uri: lb://zlt-rest-dubbo
          predicates:
            - Path=/api-rest/**
          filters:
            - StripPrefix=1

分別定義兩個路由策略：

• 路徑 /api-web/ 為請求 web-dubbo 工程
• 路徑 /api-rest/ 為請求 rest-dubbo 工程

 

4.5. 測試

分別啟動：Nacos、sc-gateway、web-dubbo、rest-dubbo 工程，通過網關的以下兩個接口分別測試兩種整合方式

1. http://127.0.0.1:9900/api-web/test/abc ：請求 web-dubbo 工程測試整合方式一
2. http://127.0.0.1:9900/api-rest/test/abc ：請求 rest-dubbo 工程測試整合方式二

 

五、demo下載

ide需要安裝 lombok 插件

https://github.com/zlt2000/dubboSpringCloud

 
掃碼關注有驚喜！

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

美國豪華電動車製造商特斯拉 (Tesla) 跨界家用電池，德國汽車碳纖維生產商西格里集團 (SGL Carbon SE) 在投資人期待該公司可望因此受惠的激勵下，股價創下近三個月以來最大單日漲幅。   SGL 主要是為日立、Panasonic 這些日本電子大廠供應石墨陽極材料，而日立、Panasonic 則會將電池元件賣給特斯拉。   Bankhaus Lampe 分析師 Marc Gabriel 說，SGL 是少數幾家能因電池需求增溫而受惠的德國業者。根據報導，SGL 發言人已確認，該公司的確是日立、Panasonic 的石墨陽極材料供應商。

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

TCP協議粘包問題詳解

前言

　　在本章節中，我們將探討TCP協議基於流式傳輸的最大一個問題，即粘包問題。本章主要介紹TCP粘包的原理與其三種解決粘包的方案。並且還會介紹為什麼UDP協議不會產生粘包。

 

基於TCP協議的socket實現遠程命令輸入

　　我們準備做一個可以在Client端遠程執行Server端shell命令並拿到其執行結果的程序，而涉及到網絡通信就必然會出現socket模塊，關於如何抉擇傳輸層協議的選擇？我們選擇使用TCP協議，因為它是可靠傳輸協議且數據量支持比UDP協議要大。好了廢話不多說直接上代碼了。

 

　　Server端代碼如下：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

# ==== 基於TCP協議的socket實現遠程命令輸入之Server ====

import subprocess
from socket import *

server = socket(AF_INET, SOCK_STREAM)
server.bind(("0.0.0.0",6666))  # 放在遠程填入0.0.0.0，放在本地填入127.0.0.1
server.listen(5)

while 1:  # 鏈接循環
    conn,client_addr = server.accept()
    while 1:  # 通信循環
        try:  # 防止Windows平台下Client端異常關閉導致雙向鏈接崩塌Server端異常的情況發生
            cmd = conn.recv(1024)
            if not cmd:  # 防止類Unix平台下Client端異常關閉導致雙向鏈接崩塌Server端異常的情況發生
                break
            res = subprocess.Popen(cmd.decode("utf-8"),
                             shell=True,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE,)

            stdout_res = res.stdout.read()  # 正確結果
            stderr_res = res.stderr.read()  # 錯誤結果
            # subprocess模塊拿到的是bytes類型，所以直接發送即可

            cmd_res = stdout_res if stdout_res else stderr_res  # 因為兩個結果只有一個有信息，所以我們只拿到有結果的那個
            conn.send(cmd_res)

        except Exception:
            break

    conn.close()  # 由於client端鏈接異常，故關閉鏈接循環

 

　　Client端代碼如下:

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

# ==== 基於TCP協議的socket實現遠程命令輸入之Client ====

from socket import *

client = socket(AF_INET,SOCK_STREAM)
client.connect(("xxx.xxx.xxx.xxx",6666))  # 填入Server端公網IP

while 1:
    cmd = input("請輸入命令>>>:").strip()
    if not cmd:
        continue
    if cmd == "quit":
        break
    client.send(cmd.encode("utf-8"))
    cmd_res = client.recv(1024)  # 本次接收1024字節數據
    print(cmd_res.decode("utf-8"))  # 如果Server端是Windows則用gbk解碼，類Unix用utf-8解碼

client.close()

 

　　測試結果：

 

粘包問題及其原理

　　上面的測試一切看起來都非常完美，但是是有一個BUG的。當我們如果讀取一條非常長的命令實際上是會出問題的，比如：

　　這種現象被稱之為粘包，那麼為何會產生這樣的現象呢？

 

　　這是由於recv()沒有一次性讀取完整個內核緩衝區的內容導致的。其實歸根結底還是怪TCP是字節流方式傳輸數據。

 

　　我們來解析一下這種現象產生的原因：

 

　　由於我們的recv()只是按照固定的1024去讀取數據，那麼一旦整體內核緩衝區中所存儲的整體數據大於1024，就會產生粘包現象。所謂粘包問題主要還是因為接收方不知道消息之間的界限，不知道一次性提取多少字節的數據所造成的。

 

　　這裏我還畫了一幅圖，可以方便讀者理解：

 

　　那麼我們可以通過不斷的增大recv()中的讀取範圍來解決這個問題嗎？就像對應上圖中的，一次性把快遞櫃包裹全取完，答案是不可以！你再大你也不可能大過內核緩衝區，這個東西都是有一個一定的閾值。一旦超出了這個閾值就會引發異常或者乾脆無效。那麼有什麼好的辦法呢？哈，下面會教給你一些解決辦法的。不過在此之前我們要先看一個TCP協議特有的Nagle算法。

 

Nagle算法與粘包

 

　　基於TCP協議的socket通信有一個特點，即：一方的send()與另一方的recv()可以沒有任何關係，即：一方send()三次，另一方recv()一次就可以將數據全部取出來。

 

　　TCP協議的發送方有一個特徵。他會進行組包，如果一次發送的數據量很小，比如第一次發送10個字節，第二次發生2個字節，第三次發生3個字節。他可能會將這15個字節湊到一塊發送出去，這是採用了Nagle算法來進行的，這麼做有一個弊端就是接收方想要將這個大的數據包按照發送方的發送次數精確無誤的接收拆分成10 2 3必須要有發送方提供的拆包機制才行。

 

　　如下圖組所示

 

　　發送方：

from socket import *
ip_port = ("127.0.0.1",12306)
buffer_size = 1024
back_log = 5

server = socket(AF_INET,SOCK_STREAM)
server.bind(ip_port)
server.listen(back_log)

conn,addr = server.accept()
conn.send("hello,".encode("utf-8"))  # 第一次發送是6Bytes的數據
conn.send("world,".encode("utf-8"))     # 第二次也是6Bytes的數據
conn.send("yunyaGG!!".encode("utf-8"))  # 第三次是9Bytes的數據

 

　　接收方：

from socket import *
ip_port = ("127.0.0.1",12306)
buffer_size = 1024

client = socket(AF_INET,SOCK_STREAM)
client.connect(ip_port)

data_1 = client.recv(buffer_size)  # 我們讀取數據時統一用設定的 buffer_size 來讀取
print("這是第一次的數據包:",data_1.decode("utf-8"))
data_2 = client.recv(buffer_size)
print("這是第二次的數據包:",data_2.decode("utf-8"))
data_3 = client.recv(buffer_size)
print("這是第三次的數據包:",data_3.decode("utf-8"))

 

　　接收結果：

# ==== 執行結果 ====
​
"""
這是第一次的數據包: hello,
這是第二次的數據包: world,yunyaGG!!
這是第三次的數據包: 
"""

 

　　和預想的有點不太一樣哈，居然把第二次和第三次組成了一個大的數據包發送過來了。這就是Nagle算法，這樣的組包策略很容易就會產生粘包。我不知道你是以什麼樣的方式發過來的，所以我recv()就只能按照自己設定的方式去接收。

 

　　現在思考一下粘包的思路，我們的發送方需要將切分解包的規則告訴給接收方。

　　我們嘗試改一下每一次的buffer_size接收大小：

 

　　接收方：

from socket import *
ip_port = ("127.0.0.1",12306)
buffer_size = 1024

client = socket(AF_INET,SOCK_STREAM)
client.connect(ip_port)

data_1 = client.recv(6)  # 我們手動的按照對方發送時的規則來進行拆包
print("這是第一次的數據包:",data_1.decode("utf-8"))
data_2 = client.recv(6)
print("這是第二次的數據包:",data_2.decode("utf-8"))
data_3 = client.recv(9)
print("這是第三次的數據包:",data_3.decode("utf-8"))

 

　　接收結果：

# ==== 執行結果 ====
​
"""
這是第一次的數據包: hello,
這是第二次的數據包: world,
這是第三次的數據包: yunyaGG!!
"""

 

　　粘包被我們手動的計算字節數來精確的分割數據接受量的大小給解決了，但是這樣做是不現實的..我們不可能知道對方發送的數據到底是怎麼樣的，更不用說手動計算。所以有沒有更好的解決方案呢？

 

解決方案1：預先發送消息長度

　　好了，其實上面關於解決粘包的思路已經出來了。我們需要做的就是讓接收方知道本次發送內容的大小，接收方才能夠精確的將所有數據全部提取出來不產生遺漏。其實實現方式很簡單，可以嘗試以下思路：

 

　　1.發送方發送一個此次數據固定的長度

　　2.接收方接收到該數據長度並且回應

　　3.發送方收到回應並且發送真正的數據

　　4.接收方不斷的用默認的buffer_size值接收新的數據並存儲起來直到超出整個數據的長度，代表此處數據全部接收完畢

 

　　Server端：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

# ==== 基於TCP協議的socket實現遠程命令輸入之Server ====

import subprocess
from socket import *

server = socket(AF_INET, SOCK_STREAM)
server.bind(("0.0.0.0", 6666))  # 放在遠程填入0.0.0.0 放在本地測試填入127.0.0.1
server.listen(5)

while 1:  # 鏈接循環
    conn, client_addr = server.accept()
    while 1:  # 通信循環
        try:  # 防止Windows平台下Client端異常關閉導致雙向鏈接崩塌Server端異常的情況發生
            cmd = conn.recv(1024)
            if not cmd:  # 防止類Unix平台下Client端異常關閉導致雙向鏈接崩塌Server端異常的情況發生
                break
            res = subprocess.Popen(cmd.decode("utf-8"),
                                   shell=True,
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE, )

            stdout_res = res.stdout.read()  # 正確結果
            stderr_res = res.stderr.read()  # 錯誤結果
            # subprocess模塊拿到的是bytes類型，所以直接發送即可

            cmd_res = stdout_res if stdout_res else stderr_res  # 因為兩個結果只有一個有信息，所以我們只拿到有結果的那個
            msg_length = len(cmd_res)  # 本次數據的長度
            conn.send(str(msg_length).encode("utf-8"))  # 先將要發的整體內容長度發送過去
            if conn.recv(1024) == b"ready":  # 如果接收方回應了ready則開始發送真正的數據體
                conn.send(cmd_res)

        except Exception:
            break

    conn.close()  # 由於client端鏈接異常，故關閉鏈接循環

 

　　Client端:

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

# ==== 基於TCP協議的socket實現遠程命令輸入之Client ====

from socket import *

client = socket(AF_INET, SOCK_STREAM)
client.connect(("xxx.xxx.xxx.xxx", 6666))  # 填入Server端公網IP

while 1:
    cmd = input("請輸入命令>>>:").strip()
    if not cmd:
        continue
    if cmd == "quit":
        break
    client.send(cmd.encode("utf-8"))

    msg_length = int(client.recv(1024).decode("utf-8"))  # 接收到此次發送內容的整體長度
    recv_length = 0  # 代表已接收的內容長度
    cmd_res = b""

    client.send(b"ready")  # 發送給Server端，代表自己已經接收到此次內容長度，可以發送真正的數據啦

    while recv_length < msg_length:
        cmd_res += client.recv(1024)  # 本次接收1024字節數據,可能是一小節數據
        recv_length += len(cmd_res)  # 添加上本次讀取的長度，當全部讀取完后應該 recv_length == msg_length

    else:
        print(cmd_res.decode("utf-8"))  # 如果Server端是Windows則用gbk解碼，類Unix用utf-8解碼

client.close()

 

　　結果如下：

 

解決方案2：json+struct方案

　　其實上面的解決方案還是有一些弊端，因為Server端是發送了2次send()，第1次發送數據整體長度，第2次發送數據內容主體，這樣其實是不太好的（Server端可能同時處理多個鏈接，所以send()次數越少越好），而且如果Server端傳的是一個文件的話那麼局限性就太強了。因為我們只能將整體的消息長度發送過去而諸如文件名，文件大小之內的信息就發送不過去。

　　所以我們需要一個更加完美的解決方案，即Server端發送一次send()就將本次的數據整體長度發送過去（還可以包括文件姓名，文件大小等信息。）

 

　　struct模塊使用介紹

 

　　struct模塊可以將其某一種數據格式序列化為固定長度的Bytes類型，其中最重要的兩個方法就是pack()、unpack()。

 

　　pack(fmt,*args): 根據格式將其轉換為Bytes類型

　　unpack(fmt,string):根據格式將Bytes類型數據反解為其原本的形式

 

格式	C語言類型	Python類型	字節數大小
x	填充字節	沒有值
c	char	字節長度為1	1
b	signed char	整數	1
B	unsigned char	整數	1
?	_Bool	bool	1
h	short	整數	2
H	unsigned short	整數	2
i	int	整數	4
I	unsigned int	整數	4
l	long	整數	4
L	unsigned long	整數	4
q	long long	整數	8
Q	unsigned long long	整數	8
n	ssize_t	整數
N	size_t	整數
f	float	浮點數	4
d	double	浮點數	8
s	char[]	字節
p	char[]	字節
P	void *	整數

 

　　使用演示：

>>> import struct
>>> b1 = struct.pack("i",12)  # 嘗試將 int類型的12進行序列化，得到一個4字節的對象
>>> b1
b'\x0c\x00\x00\x00'
>>> struct.unpack("i",b1)  # 嘗試將12的序列化對象字節進行反解，得出元組，第1位就是需要的數據。
(12,)
>>>

 

　　好了，了解到這裏我們就可以開始進行改寫了。

　　Server端代碼如下：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

# ==== 基於TCP協議的socket實現遠程命令輸入之Server ====

import json
import struct
import subprocess
from socket import *

server = socket(AF_INET, SOCK_STREAM)
server.bind(("0.0.0.0", 6666))  # 放在遠程填入0.0.0.0 放在本地測試填入127.0.0.1
server.listen(5)

while 1:  # 鏈接循環
    conn, client_addr = server.accept()
    while 1:  # 通信循環
        try:  # 防止Windows平台下Client端異常關閉導致雙向鏈接崩塌Server端異常的情況發生
            cmd = conn.recv(1024)
            if not cmd:  # 防止類Unix平台下Client端異常關閉導致雙向鏈接崩塌Server端異常的情況發生
                break
            res = subprocess.Popen(cmd.decode("utf-8"),
                                   shell=True,
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE, )

            stdout_res = res.stdout.read()  # 正確結果
            stderr_res = res.stderr.read()  # 錯誤結果
            # subprocess模塊拿到的是bytes類型，所以直接發送即可

            cmd_res = stdout_res if stdout_res else stderr_res  # 因為兩個結果只有一個有信息，所以我們只拿到有結果的那個

            # 解決粘包：構建字典,包含數據主體長度,這個就相當於其頭部信息
            head_msg = {
                "msg_length": len(cmd_res), # 包含數據主體部分的長度
                # 如果是文件，還可以添加file_name，file_size等屬性。
            }

            # 序列化成json格式，並且統計其頭部的長度
            head_data = json.dumps(head_msg).encode("utf-8")
            head_length = struct.pack("i", len(head_data))  # 得到4字節的頭部信息，裡面包含頭部的長度

            # 發送頭部長度信息，頭部數據，與真實數據部分
            conn.send(head_length + head_data + cmd_res)

        except Exception:
            break

    conn.close()  # 由於client端鏈接異常，故關閉鏈接循環

 

　　Client端代碼如下：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

# ==== 基於TCP協議的socket實現遠程命令輸入之Client ====

import json
import struct
from socket import *

client = socket(AF_INET, SOCK_STREAM)
client.connect(("xxx.xxx.xxx.xxx", 6666))  # 填入Server端公網IP

while 1:
    cmd = input("請輸入命令>>>:").strip()
    if not cmd:
        continue
    if cmd == "quit":
        break
    client.send(cmd.encode("utf-8"))  # 發送終端命令

    # 解決粘包
    head_length = struct.unpack("i", client.recv(4))[0]  # 接收到頭部的長度信息
    head_data = json.loads(client.recv(head_length))  # 接收到真實的頭部信息

    msg_length = head_data["msg_length"]  # 獲取到數據主體的長度信息
    recv_length = 0  # 代表已接收的內容長度
    cmd_res = b""

    # 開始獲取真正的數據主體信息
    while recv_length < msg_length:
        cmd_res += client.recv(1024)  # 本次接收1024字節數據,可能是一小節數據
        recv_length += len(cmd_res)  # 添加上本次讀取的長度，當全部讀取完后應該 recv_length == msg_length

    else:
        print(cmd_res.decode("utf-8"))  # 如果Server端是Windows則用gbk解碼，類Unix用utf-8解碼


client.close()

 

　　思想如下：

　　　　1.Server端構建自身的數據頭部分，其中包含數據體整體長度，如果傳輸的是文件的話還可以包含文件名，文件大小等信息

　　　　2.將數據頭部分json序列化后再轉換為Bytes類型

　　　　3.使用struct.pack()模塊獲取數據頭的長度，得到一個長度為4的Bytes類型

　　　　4.Server端將 數據頭長度 + 數據頭部分 + 數據體部分 全部發送給Client端

　　　　5. Client端recv()接收值改為4，拿到數據頭長度Bytes類型

　　　　6. Client端使用struct.unpack(數據頭長度Bytes類型)模塊反解出數據頭真實的長度

　　　　7. Client端使用recv()接收值為數據頭真實的長度拿到真正的數據頭

　　　　8. 通過json反序列化出真正的數據頭，在到其中取出數據體的長度

　　　　9. 開始while循環不斷的讀取真實的數據體數據

 

 

解決方案3：iter()與偏函數（失敗案例）

 

　　上面那麼做看似完美但還是美中不足。因為內存緩衝區本來就是只能取一次值，和迭代器很像，只能迭代一次便不能繼續迭代了。基於這一點我們來做一個終極優化：

　　還記得iter()方法嗎？iter()方法除開創建迭代器外實際上還有一個參數：

 

def iter(source, sentinel=None):  # known special case of iter
    """
    iter(iterable) -> iterator
    iter(callable, sentinel) -> iterator

    Get an iterator from an object.  In the first form, the argument must
    supply its own iterator, or be a sequence.
    In the second form, the callable is called until it returns the sentinel.
    """
    pass

 

　　我們來試試這個參數做什麼用的。

li = [1, 2, 3, 4]

def my_iter():
    return li.pop()

res = iter(my_iter, 2)  # 代表這個迭代器沒__next__一下就會執行my_iter函數，並且該函數返回值如果是2則終止迭代
print(res.__next__())  # 4
print(res.__next__())  # 3
print(res.__next__())  # StopIteration

 

　　第二個參數看來可以設置迭代的終點。

 

　　那麼偏函數是什麼呢？偏函數可以設定一個固定的參數給第一個位置的值

　　效果如下：

from functools import partial  # 導入偏函數

def add(x, y):
    return x + y

func = partial(add, 1)  # 設置辨寒暑綁定的第一個參數的值
print(func(1))  # 2
print(func(5))  # 6

 

　　現在我們仔細回想，當緩衝區的消息接收完畢後為空的狀態是會變成 b""的形式。那麼這個時候我們可以使用iter()方法設置為不斷的取出緩存中的值直到出現b""，而偏函數可以對recv()函數進行設置讓它始終取一個值，最後通過join來拼接出取出的所有值即可。

　　可以使用 "".join(iter(partial(tcp_clien.recv,back_log)),b"")

 

　　我們嘗試用函數來查看一下效果：

from functools import partial  # 導入偏函數

li = [b"","1","2","3","4","5"]  # 模擬內核緩衝區

def test(buffer_size):
    if buffer_size:  # 模擬recv的數據大小
        return li.pop()
    print("buffer_size必須為一個int類型的值")

res = "".join(iter(partial(test,1024),b""))
print(res)  # 54321

# join()方法會不斷的調用iter()下的__next__，每調用一次就執行一次偏函數。知道出現b""停止

 

　　最後我們發現，這樣的做法是會產生recv()阻塞的，總體來說還是不能夠成功。因為join()方法會不斷的執行，即使內核緩衝區的數據被recv()讀完了也不會終止迭代而是繼續阻塞下次的recv()，故這種方式宣告失敗。（還是iter()的第二個參數導致的，或許讀取完后內核緩衝區中的數據並不是b""）

 

　　測試的Server端代碼如下：

from socket import *
import subprocess
import struct
ip_port=('127.0.0.1',8080)
back_log=5
buffer_size=1024

tcp_server=socket(AF_INET,SOCK_STREAM)
tcp_server.bind(ip_port)
tcp_server.listen(back_log)

while True:
    conn,addr=tcp_server.accept()
    print('新的Client鏈接',addr)
    while True:
        #收
        try:
            cmd=conn.recv(buffer_size)
            if not cmd:break
            print('收到Client的命令',cmd)

            #執行命令，得到命令的運行結果cmd_res
            res=subprocess.Popen(cmd.decode('utf-8'),shell=True,
                                 stderr=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stdin=subprocess.PIPE)
            err=res.stderr.read()
            if err:
                cmd_res=err
            else:
                cmd_res=res.stdout.read()

            #發
            if not cmd_res:
                cmd_res='執行成功'.encode('gbk')

            length=len(cmd_res)

            data_length=struct.pack('i',length)
            conn.send(data_length)
            conn.send(cmd_res)
        except Exception as e:
            print(e)
            break

 

　　測試的Client代碼如下：

from socket import *
import struct
from functools import partial   #偏函數
ip_port=('127.0.0.1',8080)
back_log=5
buffer_size=1024

tcp_client=socket(AF_INET,SOCK_STREAM)
tcp_client.connect(ip_port)

while True:
    cmd=input('>>: ').strip()
    if not cmd:continue
    if cmd == 'quit':break

    tcp_client.send(cmd.encode('utf-8'))


    #解決粘包
    length_data=tcp_client.recv(4)
    length=struct.unpack('i',length_data)[0]
   
  #第一種方法
    recv_size=0
    recv_msg=b''
    while recv_size < length:
        #為何recv里是buffer_size，不是length，因為length如果為24G，系統內存沒有那麼大
        #所以每次buffer_size,當recv_size < length時，循環接收，直到recv_size =length，退出循環
        recv_msg += tcp_client.recv(buffer_size)
        recv_size=len(recv_msg) #1024

    #第二種方法 失敗版本，會引發recv()的阻塞，而不會終止迭代。因為join()方法會不斷的調用其iter()方法產生的迭代器，也就是調用其__next__方法，所以第二次沒消息的recv()會阻塞住。
    #recv_msg=''.join(iter(partial(tcp_client.recv, buffer_size), b''))
    print('命令的執行結果是 ',recv_msg.decode('gbk'))
tcp_client.close()

 

UDP協議為何不會產生粘包

 

　　UDP協議是面向消息的協議，每一次的sendto()與recvfrom()必須一一對應，否則就會收不到消息。

 

　　UDP是面向消息的協議，每個UDP段都是一條消息，每sendto()一次就是發送一次消息，而不管接收方有沒有收到消息發送方只管自己的發送任務，這也是UDP被稱為不可靠傳輸協議的由來。接收端的套接字緩衝區採用了鏈式的結構來記錄每一個到達的UDP包，在每一個UDP包中都有了消息頭，包括端口，消息源等等..於是UDP就能夠去區分出一個明確的消息定義，即面向消息的通信是有消息邊界的，所以UDP的傳輸叫做數據報的形式。

 

　　並且每一次recvform()的buffer_size最大值如果不夠獲取完全部的內核緩衝區里的數據的話，那麼只會收夠指定的最大字節數量(即buffer_size的設定值)，剩餘的就不要了。所以UDP不會存在粘包，多麼乾脆利落…

 

　　我們還是用一個快遞員的那個圖來進行演示：

　　還有一點需要注意一下。使用UDP協議進行通信的時候不管首先啟動哪一方都不會報錯，因為它只管發，不管有沒有人接收。

　　所以，這也是我稱UDP協議比較隨便的原因。

 

　　那麼隨便有沒有什麼好處呢？有的，速度快。不用建立雙向鏈接通道，但是其代價就是數據可靠性與安全性的問題，效率和安全從來都是相對的，這個也只能在從中做取捨。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

本文源碼：GitHub·點這裏 || GitEE·點這裏

一、流量控制

1、基本概念

流量控制的核心作用是限制流出某一網絡的某一連接的流量與突發，使這類報文以比較均勻的速度流動發送，達到保護系統相對穩定的目的。通常是將請求放入緩衝區或隊列內，然後基於特定策略處理請求，勻速或者批量處理，該過程也稱流量整形。

流量控制的核心算法有以下兩種：漏桶算法和令牌桶算法。

2、漏桶算法

基礎描述

漏桶算法是流量整形或速率限制時經常使用的一種算法，它的主要目的是控制數據注入到網絡的速率，平滑網絡上的突發流量。漏桶算法提供了一種機制，通過它，突發流量可以被整形以便為網絡提供一個穩定的流量。

漏桶算法基本思路：請求（水流）先進入到容器（漏桶）里，漏桶以一定的速度出水，這裏就是指流量流出的策略，當流量流入速度過大容器無法承接就會直接溢出，通過該過程限制數據的傳輸速率。

核心要素

通過上述流程，不難發現漏桶算法涉及下面幾個要素：

容器容量

容器的大小直接決定能承接流量的多少，容器一但接近飽和，要麼溢出，要麼加快流速；

流出速度

流量流出的速度取決於服務的請求處理能力，接口支撐的併發越高，流速就可以越大；

時間控制

基於時間記錄，判斷流量流出速度，控制勻速模式，

注意：需要一個基本的判定策略，漏桶算法在系統能承接當前併發流量時，不需要啟用。

3、令牌桶算法

基礎描述

令牌桶可自行以恆定的速率源源不斷地產生令牌。如果令牌不被消耗，或者被消耗的速度小於產生的速度，令牌就會不斷地增多，直到把桶填滿。後面再產生的令牌就會從桶中溢出。

令牌桶算法雖然根本目的也是控制流量速度，但是當令牌桶內的令牌足夠多時，則允許流量階段性的併發。傳送到令牌桶的數據包需要消耗令牌。不同大小的數據包，消耗的令牌數量不一樣。

核心要素

令牌桶

存放按照特定的速率生成的令牌，以此控制流量速度。

匹配規則

這裏的匹配規則更多是服務於分佈式系統，例如服務A是系統的核心交易，當出現併發時，基於令牌桶最匹配規則，只允許交易請求通過，例如：常見雙十一期間，各大電商平台提示，為保證核心交易，邊緣服務的數據延遲或暫停等。

注意：令牌桶算法和漏桶算法的目的雖然相同，但是實現策略是相反的，不過都存在一個問題，為保證大部分請求流量成功，會犧牲小部分請求。

二、限流組件

1、Nginx代理組件

Nginx反向代理實際運行方式是指以代理服務器來接收客戶端連接請求，然後將請求轉發給內部網絡上的服務器，並將從服務器上得到的結果返回給客戶端，此時代理服務器對外就表現為一個服務器。

流量限制是Nginx作為代理服務中一個非常實用的功能，通過配置方式來限制用戶在給定時間內HTTP請求的數量，兩個主要的配置指令limit_req_zone和limit_req，以此保護高併發下系統的穩定。

2、CDN邊緣節點

CDN邊緣節點，準確的說並不是用來處理流量限制的，而是存放靜態頁面。內容緩存為CDN網絡節點，位於用戶接入點，是面向最終用戶的內容提供設備，可緩存靜態Web內容和流媒體內容，實現內容的邊緣傳播和存儲，以便用戶的就近訪問，這樣避免用戶大量刷新數據服務器，節省骨幹網帶寬，減少帶寬需求量。

在高併發場景下，尤其是倒計時搶購類似業務，在活動開始前後用戶會產生大量刷新頁面的操作，基於CDN節點，這些請求不會下沉到數據的服務接口上。也可以基於頁面做一些請求攔截，比如點擊頁面單位時間內只放行一定量的請求，以此也可以實現一個限流控制。

三、熔斷器組件

所謂熔斷器機制，即類似電流的保險器，當然電壓過高會自動跳閘，從而保護電路系統。微服務架構中服務保護也是這個策略，當服務被判斷異常，會從服務列表斷開，等待恢復在重新連接。服務熔斷降級的策略實現有如下幾個常用的組件。

1、Hystrix組件

基礎簡介

Hystrix當前處於維護模式，即不再更新，作為SpringCloud微服務組件中，最原生的一個熔斷組件，很多思路還是有必要了解一下。例如：服務熔斷，阻止故障的連鎖反應，快速失敗並迅速恢復，服務降級等。

某個微服務發生故障時，要快速切斷服務，提示用戶，後續請求，不調用該服務，直接返回，釋放資源，這就是服務熔斷。

熔斷器策略

服務器高併發下，壓力劇增的時候,根據當業務情況以及流量，對一些服務和頁面有策略的降級(可以理解為關閉不必要的服務)，以此緩解服務器資源的壓力以保障核心任務的正常運行。熔斷生效后，會在指定的時間后調用請求來測試依賴是否恢復，依賴的應用恢復后關閉熔斷。

基本流程：

首先判斷服務熔斷器開關狀態，服務如果未熔斷則放行請求；如果服務處於熔斷中則直接返回。

每次調用都執行兩個函數markSuccess(duration)和markFailure(duration) 來統計在一定的時間段內的調用是成功和失敗次數。

基於上述的成功和失敗次數的計算策略，來判斷是否應該打開熔斷器，如果錯誤率高於一定的閾值，就會觸發熔斷機制。

熔斷器有一個生命周期，周期過後熔斷器器進入半開狀態，允許放行一個試探請求；否則，不允許放行。

2、Sentinel組件

基礎簡介

基於微服務的模式，服務和服務之間的穩定性變得越來越重要。Sentinel以流量為切入點，從流量控制、熔斷降級、系統負載保護等多個維度保護服務的穩定性。

Sentinel可以針對不同的調用關係，以不同的運行指標（如QPS、併發調用數、系統負載等）為基準，收集資源的路徑，並將這些資源的調用路徑以樹狀結構存儲起來，用於根據調用路徑對資源進行流量控制。

流量整形策略

直接拒絕模式是默認的流量控制方式，即請求超出任意規則的閾值后，新的請求就會被立即拒絕。

啟動預熱模式：當流量激增的時候，控制流量通過的速率，讓通過的流量緩慢增加，在一定時間內逐漸增加到閾值上限，給冷系統一個預熱的時間，避免冷系統被壓垮。

勻速排隊方式會嚴格控制請求通過的間隔時間，也即是讓請求以均勻的速度通過，對應的是漏桶算法。

熔斷策略

Sentinel本質上是基於熔斷器模式，支持基於異常比率的熔斷降級，在調用達到一定量級並且失敗比率達到設定的閾值時自動進行熔斷，此時所有對該資源的調用都會被阻塞，直到過了指定的時間窗口后才啟發性地恢復。

四、源代碼地址

GitHub·地址
https://github.com/cicadasmile/husky-spring-cloud
GitEE·地址
https://gitee.com/cicadasmile/husky-spring-cloud

推薦閱讀：微服務架構系列

序號	標題
01	微服務架構：項目技術選型簡介，架構圖解說明
02	微服務架構：業務架構設計，系統分層管理
03	微服務架構：數據庫選型簡介，業務數據規劃設計
04	微服務架構：中間件集成，公共服務封裝
05	微服務架構：SpringCloud 基礎組件應用設計
06	微服務架構：通過業務、應用、技術、存儲，聊聊架構
07	微服務技術棧：常見註冊中心組件，對比分析

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?