一、前言

從 18 年開始接觸 .NET Core 開始，在私底下、工作中也開始慢慢從傳統的 mvc 前後端一把梭，開始轉向 web api + vue，之前自己有個半成品的 asp.net core 2.2 的項目模板，最近幾個月的時間，私下除了學習 Angular 也在對這個模板基於 asp.net core 3.1 進行慢慢補齊功能

因為涉及到底層框架大版本升級，由於某些 breaking changes 必定會造成之前的某些寫法沒辦法繼續使用，趁着端午節假期，在改造模板時，發現沒辦法通過構造函數注入的形式在 Startup 文件中注入某些我需要的服務了，因此本篇文章主要介紹如何在 asp.net core 3.x 的 startup 文件中獲取注入的服務

二、Step by Step

2.1、問題案例

這個問題的發現源於我需要改造模型驗證失敗時返回的錯誤信息，如果你有嘗試的話，在 3.x 版本中你會發現在 Startup 類中，我們沒辦法通過構造函數注入的方式再注入任何其它的服務了，這裏僅以我的代碼中需要解決的這個問題作為案例

在定義接口時，為了降低後期調整的複雜度，在接收參數時，一般會將參數包裝成一個 dto 對象（data transfer object – 數據傳輸對象），不管是提交數據，還是查詢數據，對於這個 dto 中的某些屬性，都會存在一定的卡控，例如 xxx 字段不能為空了，xxx 字段的長度不能超過 30

而在 asp.net core 中，因為會自動進行模型驗證，當不符合 dto 中的屬性要求時，接口會自動返回錯誤信息，默認的返回信息如下圖所示

可以看到，因為這裏其實是按照 rfc7231這個 RFC 協議返回的錯誤信息，這個並不符合我的要求，因此這裏我需要改寫這個返回的錯誤信息

自定義 asp.net core 的模型驗證錯誤信息方法有很多種，我的實現方法如下，因為我需要記錄請求的標識 Id 和錯誤日誌，所以這裏我需要將 ILogger 和 IHttpContextAccessor 注入到 Startup 類中

/// <summary>
/// 修改模型驗證錯誤返回信息
/// </summary>
/// <param name="services">服務容器集合</param>
/// <param name="logger">日誌記錄實例</param>
/// <param name="httpContextAccessor"></param>
/// <returns></returns>
public static IServiceCollection AddCustomInvalidModelState(this IServiceCollection services,
    ILogger<Startup> logger, IHttpContextAccessor httpContextAccessor)
{
    services.Configure<ApiBehaviorOptions>(options =>
    {
        options.InvalidModelStateResponseFactory = actionContext =>
        {
            // 獲取驗證不通過的字段信息
            //
            var errors = actionContext.ModelState.Where(e => e.Value.Errors.Count > 0)
                .Select(e => new ApiErrorDto
                {
                    Title = "請求參數不符合字段格式要求",
                    Message = e.Value.Errors.FirstOrDefault()?.ErrorMessage
                }).ToList();

            var result = new ApiReturnDto<object>
            {
                TraceId = httpContextAccessor.HttpContext.TraceIdentifier,
                Status = false,
                Error = errors
            };

            logger.LogError($"接口請求參數格式錯誤: {JsonConvert.SerializeObject(result)}");

            return new BadRequestObjectResult(result);
        };
    });

    return services;
}

在 asp.net core 2.x 版本中，你完全可以像在別的類中採用構造函數注入的方式一樣直接注入使用

public class Startup
{
    /// <summary>
    /// 日誌記錄實例
    /// </summary>
    private readonly ILogger<Startup> _logger;

    /// <summary>
    /// Http 請求實例
    /// </summary>
    private readonly IHttpContextAccessor _httpContextAccessor;

    /// <summary>
    /// ctor
    /// </summary>
    /// <param name="configuration"></param>
    /// <param name="logger"></param>
    /// <param name="httpContextAccessor"></param>
    public Startup(IConfiguration configuration, ILogger<Startup> logger, IHttpContextAccessor httpContextAccessor)
    {
        Configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _httpContextAccessor = httpContextAccessor ?? throw new ArgumentNullException(nameof(httpContextAccessor));
    }

    /// <summary>
    /// 配置實例
    /// </summary>
    public IConfiguration Configuration { get; }

    /// <summary>
    /// This method gets called by the runtime. Use this method to add services to the container.
    /// </summary>
    public void ConfigureServices(IServiceCollection services)
    {
        //注入的其它服務

        // 返回自定義的模型驗證錯誤信息
        services.AddCustomInvalidModelState(_logger, _httpContextAccessor);
    }
}

但是當你直接遷移到 asp.net core 3.x 版本后，你會發現程序會報如下的錯誤，很常見的一個依賴注入的錯誤，源頭直指我們通過構造函數注入的 ILogger、IHttpContextAccessor 接口

2.2、解決方法

根本原因

通過查閱 stackoverflow 發現了這樣的一個問題：How do I write logs from within Startup.cs，在最高贊的回答中提到了在泛型主機（GenericHostBuilder）中，沒辦法注入除 IConfiguration 之外的任何服務到 Startup類中，而泛型主機則是在 asp.net core 3.0 中添加的功能

查了下升級日誌，從中可以看到，在泛型主機中， Startup 類的構造函數注入只支持 IHostEnvironment、 IWebHostEnvironment、IConfiguration ，嗯，不好好看別人文檔的鍋

為什麼使用 WebHostBuilder可以，換成 GenericHostBuilder 就不行了呢

按照正常的邏輯來說，對於一個 asp.net core 應用，原則上來說只有有一個根級（root）的依賴注入容器，但是因為我們在 Startup 類中通過構造函數注入的形式注入服務時，告訴程序了我需要這個服務的實例，從而導致在構建 WebHost 時存在了一個單獨的容器，並且這個容器只包含了我們需要使用到的服務信息，之後，因為會創建了一個包含完整服務的依賴注入容器，這裏就會存在一個服務哪怕是單例的也可能會存在註冊兩次的問題，這無疑有些不太合乎規範

在推行泛型主機之後，嚴格控制了只會存在一個依賴注入容器，而所有的服務都是在 Startup.ConfigureServices 方法執行完成后才會註冊到依賴注入容器中，因此沒辦法像之前一樣在根容器註冊完成之前通過構造函數注入的形式使用

解決方案

如果你需要在 Startup.Configure 方法中使用自定義的服務，因為這裏已經完成了各種服務的註冊，和之前一樣，我們直接在方法簽名中包含需要使用到的服務即可

public void Configure(IApplicationBuilder app, IHostEnvironment env, ILogger<Startup> logger)
{
    logger.LogInformation("在 Configure 中使用自定義的服務");
}

如果你需要在 Startup.ConfigureServices 中使用的話，則需要換一種方法

最簡單的方法，直接替換泛型主機為原來的 WebHostBuilder，這樣就可以直接在 Startup 類中注入各種服務接口了，不過，考慮到這一改動其實是在開倒車，所以這裏不推薦採用這種方法

既然沒辦法正向通過依賴注入容器來自動創建我們需要的服務實例，是不是可以通過服務容器，手動去獲取我們需要的服務，也就是被稱為服務定位（Service Locator）的方式來獲取實例

當然，這似乎與依賴注入的思想相左，對於依賴注入來說，我們將所有需要使用的服務定義好，在應用啟動前完成註冊，之後在使用時由依賴注入容器提供服務的實例即可，而服務定位則是我們已經知道存在這個服務了，從容器中獲取出來然後由自己手動的創建實例

雖然服務定位是一種反模式，但是在某些情況下，我們又不得不採用

這裏對於本篇文章開篇中需要解決的問題，我也是採用服務定位的方式，通過構建一個 ServiceProvider 之後，手動的從容器中獲取需要使用的服務實例，調整后的代碼如下

/// <summary>
/// 添加自定義模型驗證失敗時返回的錯誤信息
/// </summary>
/// <param name="services">服務容器集合</param>
/// <returns></returns>
public static IServiceCollection AddCustomInvalidModelState(this IServiceCollection services)
{
    // 構建一個服務的提供程序
    var provider = services.BuildServiceProvider();

    // 獲取需要使用的服務實例
    //
    var logger = provider.GetRequiredService<ILogger<Startup>>();
    var httpContextAccessor = provider.GetRequiredService<IHttpContextAccessor>();

    services.Configure<ApiBehaviorOptions>(options =>
    {
        options.InvalidModelStateResponseFactory = actionContext =>
        {
            // 獲取失敗信息
            //
            var errors = actionContext.ModelState.Where(e => e.Value.Errors.Count > 0)
                .Select(e => new ApiErrorMessageDto
                {
                    Title = "Request parameters do not meet the field requirements",
                    Message = e.Value.Errors.FirstOrDefault()?.ErrorMessage
                }).ToList();

            var result = new ApiResponseDto<object>
            {
                TraceId = httpContextAccessor.HttpContext.TraceIdentifier,
                Status = false,
                Error = errors
            };

            logger.LogError($"接口請求參數格式錯誤: {JsonSerializer.Serialize(result)}");

            return new BadRequestObjectResult(result);
        };
    });

    return services;
}

對於配置一些需要基於某些服務的服務，這裏也可以通過委託的形式獲取到需要使用的服務實例，示例代碼如下

public void ConfigureServices(IServiceCollection services)
{
    services.AddSingleton<IMyService>((container) =>
    {
        var logger = container.GetRequiredService<ILogger<MyService>>();
        return new MyService
        {
            Logger = logger
        };
    });
}

三、參考資料

• ASP.NET Core 3.0 的新增功能

• Generic Host restricts Startup constructor injection

• 依賴注入模式

• Avoiding Startup service injection in ASP.NET Core 3

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

【其他文章推薦】

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

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

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

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

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

前文傳送門：

「Python 圖像處理 OpenCV （1）：入門」

「Python 圖像處理 OpenCV （2）：像素處理與 Numpy 操作以及 Matplotlib 显示圖像」

「Python 圖像處理 OpenCV （3）：圖像屬性、圖像感興趣 ROI 區域及通道處理」

「Python 圖像處理 OpenCV （4）：圖像算數運算以及修改顏色空間」

「Python 圖像處理 OpenCV （5）：圖像的幾何變換」

「Python 圖像處理 OpenCV （6）：圖像的閾值處理」

「Python 圖像處理 OpenCV （7）：圖像平滑（濾波）處理」

「Python 圖像處理 OpenCV （8）：圖像腐蝕與圖像膨脹」

「Python 圖像處理 OpenCV （9）：圖像處理形態學開運算、閉運算以及梯度運算」

「Python 圖像處理 OpenCV （10）：圖像處理形態學之頂帽運算與黑帽運算」

「Python 圖像處理 OpenCV （11）：Canny 算子邊緣檢測技術」

引言

前文介紹了 Canny 算子邊緣檢測，本篇繼續介紹 Roberts 算子、 Prewitt 算子、 Sobel 算子和 Laplacian 算子等常用邊緣檢測技術。

Roberts 算子

Roberts 算子，又稱羅伯茨算子，是一種最簡單的算子，是一種利用局部差分算子尋找邊緣的算子。他採用對角線方向相鄰兩象素之差近似梯度幅值檢測邊緣。檢測垂直邊緣的效果好於斜向邊緣，定位精度高，對噪聲敏感，無法抑制噪聲的影響。

1963年， Roberts 提出了這種尋找邊緣的算子。 Roberts 邊緣算子是一個 2×2 的模版，採用的是對角方向相鄰的兩個像素之差。

Roberts 算子的模板分為水平方向和垂直方向，如下所示，從其模板可以看出， Roberts 算子能較好的增強正負 45 度的圖像邊緣。

\[dx = \left[ \begin{matrix} -1 & 0\\ 0 & 1 \\ \end{matrix} \right] \]

\[dy = \left[ \begin{matrix} 0 & -1\\ 1 & 0 \\ \end{matrix} \right] \]

Roberts 算子在水平方向和垂直方向的計算公式如下：

\[d_x(i, j) = f(i + 1, j + 1) – f(i, j) \]

\[d_y(i, j) = f(i, j + 1) – f(i + 1, j) \]

Roberts 算子像素的最終計算公式如下：

\[S = \sqrt{d_x(i, j)^2 + d_y(i, j)^2} \]

今天的公式都是小學生水平，千萬別再說看不懂了。

實現 Roberts 算子，我們主要通過 OpenCV 中的 filter2D() 這個函數，這個函數的主要功能是通過卷積核實現對圖像的卷積運算：

def filter2D(src, ddepth, kernel, dst=None, anchor=None, delta=None, borderType=None)

• src: 輸入圖像
• ddepth: 目標圖像所需的深度
• kernel: 卷積核

接下來開始寫代碼，首先是圖像的讀取，並把這個圖像轉化成灰度圖像，這個沒啥好說的：

# 讀取圖像
img = cv.imread('maliao.jpg', cv.COLOR_BGR2GRAY)
rgb_img = cv.cvtColor(img, cv.COLOR_BGR2RGB)

# 灰度化處理圖像
grayImage = cv.cvtColor(img, cv.COLOR_BGR2GRAY)

然後是使用 Numpy 構建卷積核，並對灰度圖像在 x 和 y 的方向上做一次卷積運算：

# Roberts 算子
kernelx = np.array([[-1, 0], [0, 1]], dtype=int)
kernely = np.array([[0, -1], [1, 0]], dtype=int)

x = cv.filter2D(grayImage, cv.CV_16S, kernelx)
y = cv.filter2D(grayImage, cv.CV_16S, kernely)

注意：在進行了 Roberts 算子處理之後，還需要調用convertScaleAbs()函數計算絕對值，並將圖像轉換為8位圖進行显示，然後才能進行圖像融合：

# 轉 uint8 ,圖像融合
absX = cv.convertScaleAbs(x)
absY = cv.convertScaleAbs(y)
Roberts = cv.addWeighted(absX, 0.5, absY, 0.5, 0)

最後是通過 pyplot 將圖像显示出來：

# 显示圖形
titles = ['原始圖像', 'Roberts算子']
images = [rgb_img, Roberts]

for i in range(2):
    plt.subplot(1, 2, i + 1), plt.imshow(images[i], 'gray')
    plt.title(titles[i])
    plt.xticks([]), plt.yticks([])
plt.show()

最終結果如下：

Prewitt 算子

Prewitt 算子是一種一階微分算子的邊緣檢測，利用像素點上下、左右鄰點的灰度差，在邊緣處達到極值檢測邊緣，去掉部分偽邊緣，對噪聲具有平滑作用。

由於 Prewitt 算子採用 3 * 3 模板對區域內的像素值進行計算，而 Robert 算子的模板為 2 * 2 ，故 Prewitt 算子的邊緣檢測結果在水平方向和垂直方向均比 Robert 算子更加明顯。Prewitt算子適合用來識別噪聲較多、灰度漸變的圖像。

Prewitt 算子的模版如下：

\[dx = \left[ \begin{matrix} 1 & 0 & -1\\ 1 & 0 & -1\\ 1 & 0 & -1\\ \end{matrix} \right] \]

\[dy = \left[ \begin{matrix} -1 & -1 & -1\\ 0 & 0 & 0\\ 1 & 1 & 1\\ \end{matrix} \right] \]

在代碼實現上， Prewitt 算子的實現過程與 Roberts 算子比較相似，我就不多介紹，直接貼代碼了：

import cv2 as cv
import numpy as np
import matplotlib.pyplot as plt

# 讀取圖像
img = cv.imread('maliao.jpg', cv.COLOR_BGR2GRAY)
rgb_img = cv.cvtColor(img, cv.COLOR_BGR2RGB)

# 灰度化處理圖像
grayImage = cv.cvtColor(img, cv.COLOR_BGR2GRAY)

# Prewitt 算子
kernelx = np.array([[1,1,1],[0,0,0],[-1,-1,-1]],dtype=int)
kernely = np.array([[-1,0,1],[-1,0,1],[-1,0,1]],dtype=int)

x = cv.filter2D(grayImage, cv.CV_16S, kernelx)
y = cv.filter2D(grayImage, cv.CV_16S, kernely)

# 轉 uint8 ,圖像融合
absX = cv.convertScaleAbs(x)
absY = cv.convertScaleAbs(y)
Prewitt = cv.addWeighted(absX, 0.5, absY, 0.5, 0)

# 用來正常显示中文標籤
plt.rcParams['font.sans-serif'] = ['SimHei']

# 显示圖形
titles = ['原始圖像', 'Prewitt 算子']
images = [rgb_img, Prewitt]

for i in range(2):
    plt.subplot(1, 2, i + 1), plt.imshow(images[i], 'gray')
    plt.title(titles[i])
    plt.xticks([]), plt.yticks([])
plt.show()

從結果上來看， Prewitt 算子圖像銳化提取的邊緣輪廓，其效果圖的邊緣檢測結果比 Robert 算子更加明顯。

Sobel 算子

Sobel 算子的中文名稱是索貝爾算子，是一種用於邊緣檢測的離散微分算子，它結合了高斯平滑和微分求導。

Sobel 算子在 Prewitt 算子的基礎上增加了權重的概念，認為相鄰點的距離遠近對當前像素點的影響是不同的，距離越近的像素點對應當前像素的影響越大，從而實現圖像銳化並突出邊緣輪廓。

算法模版如下：

\[dx = \left[ \begin{matrix} 1 & 0 & -1\\ 2 & 0 & -2\\ 1 & 0 & -1\\ \end{matrix} \right] \]

\[dy = \left[ \begin{matrix} -1 & -2 & -1\\ 0 & 0 & 0\\ 1 & 2 & 1\\ \end{matrix} \right] \]

Sobel 算子根據像素點上下、左右鄰點灰度加權差，在邊緣處達到極值這一現象檢測邊緣。對噪聲具有平滑作用，提供較為精確的邊緣方向信息。因為 Sobel 算子結合了高斯平滑和微分求導（分化），因此結果會具有更多的抗噪性，當對精度要求不是很高時， Sobel 算子是一種較為常用的邊緣檢測方法。

Sobel 算子近似梯度的大小的計算公式如下：

\[G = \sqrt{d_X^2 + d_y^2} \]

梯度方向的計算公式如下：

\[\theta = \tan^{-1}(\frac {d_x}{d_y}) \]

如果以上的角度 θ 等於零，即代表圖像該處擁有縱向邊緣，左方較右方暗。

在 Python 中，為我們提供了 Sobel() 函數進行運算，整體處理過程和前面的類似，代碼如下：

import cv2 as cv
import matplotlib.pyplot as plt

# 讀取圖像
img = cv.imread('maliao.jpg', cv.COLOR_BGR2GRAY)
rgb_img = cv.cvtColor(img, cv.COLOR_BGR2RGB)

# 灰度化處理圖像
grayImage = cv.cvtColor(img, cv.COLOR_BGR2GRAY)

# Sobel 算子
x = cv.Sobel(grayImage, cv.CV_16S, 1, 0)
y = cv.Sobel(grayImage, cv.CV_16S, 0, 1)

# 轉 uint8 ,圖像融合
absX = cv.convertScaleAbs(x)
absY = cv.convertScaleAbs(y)
Sobel = cv.addWeighted(absX, 0.5, absY, 0.5, 0)

# 用來正常显示中文標籤
plt.rcParams['font.sans-serif'] = ['SimHei']

# 显示圖形
titles = ['原始圖像', 'Sobel 算子']
images = [rgb_img, Sobel]

for i in range(2):
    plt.subplot(1, 2, i + 1), plt.imshow(images[i], 'gray')
    plt.title(titles[i])
    plt.xticks([]), plt.yticks([])
plt.show()

Laplacian 算子

拉普拉斯（ Laplacian ）算子是 n 維歐幾里德空間中的一個二階微分算子，常用於圖像增強領域和邊緣提取。

Laplacian 算子的核心思想：判斷圖像中心像素灰度值與它周圍其他像素的灰度值，如果中心像素的灰度更高，則提升中心像素的灰度；反之降低中心像素的灰度，從而實現圖像銳化操作。

在實現過程中， Laplacian 算子通過對鄰域中心像素的四方向或八方向求梯度，再將梯度相加起來判斷中心像素灰度與鄰域內其他像素灰度的關係，最後通過梯度運算的結果對像素灰度進行調整。

Laplacian 算子分為四鄰域和八鄰域，四鄰域是對鄰域中心像素的四方向求梯度，八鄰域是對八方向求梯度。

四鄰域模板如下：

\[H = \left[ \begin{matrix} 0 & -1 & 0\\ -1 & 4 & -1\\ 0 & -1 & 0\\ \end{matrix} \right] \]

八鄰域模板如下：

\[H = \left[ \begin{matrix} -1 & -1 & -1\\ -1 & 4 & -1\\ -1 & -1 & -1\\ \end{matrix} \right] \]

通過模板可以發現，當鄰域內像素灰度相同時，模板的卷積運算結果為0；當中心像素灰度高於鄰域內其他像素的平均灰度時，模板的卷積運算結果為正數；當中心像素的灰度低於鄰域內其他像素的平均灰度時，模板的卷積為負數。對卷積運算的結果用適當的衰弱因子處理並加在原中心像素上，就可以實現圖像的銳化處理。

在 OpenCV 中， Laplacian 算子被封裝在 Laplacian() 函數中，其主要是利用Sobel算子的運算，通過加上 Sobel 算子運算出的圖像 x 方向和 y 方向上的導數，得到輸入圖像的圖像銳化結果。

import cv2 as cv
import matplotlib.pyplot as plt

# 讀取圖像
img = cv.imread('maliao.jpg', cv.COLOR_BGR2GRAY)
rgb_img = cv.cvtColor(img, cv.COLOR_BGR2RGB)

# 灰度化處理圖像
grayImage = cv.cvtColor(img, cv.COLOR_BGR2GRAY)

# Laplacian
dst = cv.Laplacian(grayImage, cv.CV_16S, ksize = 3)
Laplacian = cv.convertScaleAbs(dst)

# 用來正常显示中文標籤
plt.rcParams['font.sans-serif'] = ['SimHei']

# 显示圖形
titles = ['原始圖像', 'Laplacian 算子']
images = [rgb_img, Laplacian]

for i in range(2):
    plt.subplot(1, 2, i + 1), plt.imshow(images[i], 'gray')
    plt.title(titles[i])
    plt.xticks([]), plt.yticks([])
plt.show()

最後

邊緣檢測算法主要是基於圖像強度的一階和二階導數，但導數通常對噪聲很敏感，因此需要採用濾波器來過濾噪聲，並調用圖像增強或閾值化算法進行處理，最後再進行邊緣檢測。

最後我先使用高斯濾波去噪之後，再進行邊緣檢測：

import cv2 as cv
import numpy as np
import matplotlib.pyplot as plt

# 讀取圖像
img = cv.imread('maliao.jpg')
rgb_img = cv.cvtColor(img, cv.COLOR_BGR2RGB)

# 灰度化處理圖像
gray_image = cv.cvtColor(img, cv.COLOR_BGR2GRAY)

# 高斯濾波
gaussian_blur = cv.GaussianBlur(gray_image, (3, 3), 0)

# Roberts 算子
kernelx = np.array([[-1, 0], [0, 1]], dtype = int)
kernely = np.array([[0, -1], [1, 0]], dtype = int)
x = cv.filter2D(gaussian_blur, cv.CV_16S, kernelx)
y = cv.filter2D(gaussian_blur, cv.CV_16S, kernely)
absX = cv.convertScaleAbs(x)
absY = cv.convertScaleAbs(y)
Roberts = cv.addWeighted(absX, 0.5, absY, 0.5, 0)

# Prewitt 算子
kernelx = np.array([[1, 1, 1], [0, 0, 0], [-1, -1, -1]], dtype=int)
kernely = np.array([[-1, 0, 1], [-1, 0, 1], [-1, 0, 1]], dtype=int)
x = cv.filter2D(gaussian_blur, cv.CV_16S, kernelx)
y = cv.filter2D(gaussian_blur, cv.CV_16S, kernely)
absX = cv.convertScaleAbs(x)
absY = cv.convertScaleAbs(y)
Prewitt = cv.addWeighted(absX, 0.5, absY, 0.5, 0)

# Sobel 算子
x = cv.Sobel(gaussian_blur, cv.CV_16S, 1, 0)
y = cv.Sobel(gaussian_blur, cv.CV_16S, 0, 1)
absX = cv.convertScaleAbs(x)
absY = cv.convertScaleAbs(y)
Sobel = cv.addWeighted(absX, 0.5, absY, 0.5, 0)

# 拉普拉斯算法
dst = cv.Laplacian(gaussian_blur, cv.CV_16S, ksize = 3)
Laplacian = cv.convertScaleAbs(dst)

# 展示圖像
titles = ['Source Image', 'Gaussian Image', 'Roberts Image',
          'Prewitt Image','Sobel Image', 'Laplacian Image']
images = [rgb_img, gaussian_blur, Roberts, Prewitt, Sobel, Laplacian]
for i in np.arange(6):
   plt.subplot(2, 3, i+1), plt.imshow(images[i], 'gray')
   plt.title(titles[i])
   plt.xticks([]), plt.yticks([])
plt.show()

示例代碼

如果有需要獲取源碼的同學可以在公眾號回復「OpenCV」進行獲取。

參考

https://blog.csdn.net/Eastmount/article/details/89001702

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

【其他文章推薦】

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

※廣告預算用在刀口上，網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

※南投搬家前需注意的眉眉角角，別等搬了再說!

※新北清潔公司,居家、辦公、裝潢細清專業服務

struct sdshdr{
 //記錄buf數組中已使用字節的數量
 //等於 SDS 保存字符串的長度
 int len；
 //記錄 buf 數組中未使用字節的數量
 int free；
 //字節數組，用於保存字符串
 char buf[]；
}

 流程：
A預發送消息給B，然後執行A業務邏輯，B存儲預發送消息，A執行完業務邏輯發送業務操作結果給B，B更新預發送消息為確認併發送消息狀態同時發送消息給mq，然後被E監聽然後發送給F消費掉
C：對預發送消息異常的處理，去查詢待確認狀態超時的消息，去A中查詢進行數據處理，如果A中業務處理成功了，那麼C需改消息狀態為確認併發送狀態，然後發送消息給mq；如果A中業務處理失敗了..那麼C直接把消息刪除即可.
C2 ： 查詢消息的頁面，對消息的可視化，以及批量處理死亡消息；
D： B給mq放入數據如果失敗，，通過D去重試，多次重試失敗，消息設置為死亡 
E：確保F執行完成，發送消息給B刪除消息
優化建議： 
 （1）數據庫：如果用redis，持久化要配置成appendfsync always，確保每次新添加消息都能持久化進磁盤
 （2）在被動方應用業務冪等性判斷比較麻煩或者比較耗性能情況下，增加消息日誌記錄表.用於判斷之前有無發送過；

Set<Integer> ints = Set.of（1， 2， 3）；
List<String> strings = List.of（"first"， "second"）；

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

【其他文章推薦】

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!!

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!

※幫你省時又省力,新北清潔一流服務好口碑

※別再煩惱如何寫文案,掌握八大原則!

在分佈式系統中，負載均衡是必不可少的一個模塊，dubbo 中提供了五種負載均衡的實現，在閱讀這塊源碼之前，建議先學習負載均衡的基礎知識。把看源碼當做一個印證自己心中所想的過程，這樣會得到事半功倍的效果

以下源碼分析基於 dubbo 2.77 版本

類結構

先來看一下這一塊的類結構圖

大部分算法都是在權重比的基礎上進行負載均衡，RandomLoadBalance 是默認的算法

AbstractLoadBalance

AbstractLoadBalance 對一些通用的操作做了處理，是一個典型的模板方法模式的實現

select 方法只做一些簡單的範圍校驗，具體的實現有子類通過 doSelect 方法去實現

    @Override
    public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        if (CollectionUtils.isEmpty(invokers)) {
            return null;
        }
        if (invokers.size() == 1) {
            return invokers.get(0);
        }
        return doSelect(invokers, url, invocation);
    }

getWeight方法封裝了獲取一個調用者的權重值的方法，並加入了預熱處理

    int getWeight(Invoker<?> invoker, Invocation invocation) {
        int weight;
        URL url = invoker.getUrl();
        // Multiple registry scenario, load balance among multiple registries.
        // 註冊中心不需要預熱
        if (REGISTRY_SERVICE_REFERENCE_PATH.equals(url.getServiceInterface())) {
            weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT);
        } else {
            // 獲取配置的權重值
            weight = url.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);
            if (weight > 0) {
                // 獲取服務提供者啟動時的時間戳
                long timestamp = invoker.getUrl().getParameter(TIMESTAMP_KEY, 0L);
                if (timestamp > 0L) {
                    //  獲取啟動時長
                    long uptime = System.currentTimeMillis() - timestamp;
                    // 當前時間小於服務提供者啟動時間，直接給一個最小權重1
                    if (uptime < 0) {
                        return 1;
                    }
                    // 獲取預熱時間
                    int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP);
                    // 如果小於預熱時間，計算權重
                    if (uptime > 0 && uptime < warmup) {
                        weight = calculateWarmupWeight((int)uptime, warmup, weight);
                    }
                }
            }
        }
        // 取與零比較的最大值，保證不會出現負值權重
        return Math.max(weight, 0);
    }

calculateWarmupWeight 方法用來計算權重，保證隨着預熱時間的增加，權重逐漸達到設置的權重

    static int calculateWarmupWeight(int uptime, int warmup, int weight) {
        // 運行時間/(預熱時間/權重)
        int ww = (int) ( uptime / ((float) warmup / weight));
        // 保證計算的權重最小值是1，並且不能超過設置的權重
        return ww < 1 ? 1 : (Math.min(ww, weight));
    }

RandomLoadBalance

隨機調用是負載均衡算法中最常用的算法之一，也是 dubbo 的默認負載均衡算法，實現起來也較為簡單
隨機調用的缺點是在調用量比較少的情況下，有可能出現不均勻的情況

	@Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // Every invoker has the same weight?
        boolean sameWeight = true;
        // the weight of every invokers
        int[] weights = new int[length];
        // the first invoker's weight
        int firstWeight = getWeight(invokers.get(0), invocation);
        weights[0] = firstWeight;
        // The sum of weights
        int totalWeight = firstWeight;
        for (int i = 1; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            // save for later use
            // 依次把權重放到數組對應的位置
            weights[i] = weight;
            // Sum
            // 累加權重
            totalWeight += weight;
            // 如果出現權重不一樣的，sameWeight 設為false
            if (sameWeight && weight != firstWeight) {
                sameWeight = false;
            }
        }
        if (totalWeight > 0 && !sameWeight) {
            // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
            // 在總權重裏面隨機選擇一個偏移量
            int offset = ThreadLocalRandom.current().nextInt(totalWeight);
            // Return a invoker based on the random value.
            for (int i = 0; i < length; i++) {
                offset -= weights[i];
                // 依次用偏移量減去當前權重，小於0說明選中
                if (offset < 0) {
                    return invokers.get(i);
                }
            }
        }
        // If all invokers have the same weight value or totalWeight=0, return evenly.
        // 如果所有的調用者有同樣的權重或者總權重為0，則隨機選擇一個
        return invokers.get(ThreadLocalRandom.current().nextInt(length));
    }

RoundRobinLoadBalance

輪訓算法避免了隨機算法在小數據量產生的不均勻問題，我個人認為，輪訓算法可以理解為隨機算法的一種特例，在大量請求的情況下，從調用次數看，和隨機並無區別，主要區別在於短時間內的調用分配上

加權輪訓算法給人的直觀感受，實現起來並不複雜，算出一權重總量，依次調用即可
例如A，B，C 三個節點的權重比依次 1，200，1000，如果依次輪訓調用，就會出現先調用A 10 次，再調用B 200次，最後調用 C 1000次，不斷重複前面的過程
但這樣有一個問題，我們可以發現C 被練習調用1000次，會對C瞬間造成很大的壓力

dubbo的新版本採用的是平滑加權輪詢算法，輪訓的過程中節點之間穿插調用，可以避免了上面說的問題，因此這塊源碼看起來會稍有難度

輪訓算法 在dubbo 在升級的過程中，做過多次優化，有興趣的可以去了解下該算法的優化過程，也是件很有意思的事情

public class RoundRobinLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "roundrobin";

    private static final int RECYCLE_PERIOD = 60000;

    protected static class WeightedRoundRobin {
        // 權重值
        private int weight;
        // 當前權重值
        private AtomicLong current = new AtomicLong(0);
        // 最後一次使用該對象時間
        private long lastUpdate;

        public int getWeight() {
            return weight;
        }

        public void setWeight(int weight) {
            this.weight = weight;
            current.set(0);
        }

        // 獲取自增權重基數的當前權重值
        public long increaseCurrent() {
            return current.addAndGet(weight);
        }

        public void sel(int total) {
            current.addAndGet(-1 * total);
        }

        public long getLastUpdate() {
            return lastUpdate;
        }

        // 設置最後一次更新時間戳
        public void setLastUpdate(long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }
    }

    private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();

    /**
     * get invoker addr list cached for specified invocation
     * <p>
     * <b>for unit test only</b>
     *
     * @param invokers
     * @param invocation
     * @return
     */
    protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
        if (map != null) {
            return map.keySet();
        }
        return null;
    }

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // {group}/{interfaceName}:{version} + methoName 獲取當前消費者的唯一標示
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        // 獲取對應的 WeightedRoundRobin map,如果不存在，new 一個map放進去
        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
        int totalWeight = 0;
        long maxCurrent = Long.MIN_VALUE;
        long now = System.currentTimeMillis();
        Invoker<T> selectedInvoker = null;
        WeightedRoundRobin selectedWRR = null;
        for (Invoker<T> invoker : invokers) {
            // 服務提供者在的唯一標識
            String identifyString = invoker.getUrl().toIdentityString();
            int weight = getWeight(invoker, invocation);
            WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {
                WeightedRoundRobin wrr = new WeightedRoundRobin();
                wrr.setWeight(weight);
                return wrr;
            });
            // 如果權重改變了，更新 weightedRoundRobin 裏面權重的值
            if (weight != weightedRoundRobin.getWeight()) {
                //weight changed
                weightedRoundRobin.setWeight(weight);
            }
            // 當前權重自增自身權重
            long cur = weightedRoundRobin.increaseCurrent();
            // 設置最後一次更新時間戳
            weightedRoundRobin.setLastUpdate(now);
            // 如果當前權重大於最大當前權重
            if (cur > maxCurrent) {
                // 重置最大當前權重的值
                maxCurrent = cur;
                // 把當前提供者設為選中的提供者
                selectedInvoker = invoker;
                // 把當前輪訓權重實例設為選中
                selectedWRR = weightedRoundRobin;
            }
            // 累計總權重
            totalWeight += weight;
        }
        // 提供者有變化
        if (invokers.size() != map.size()) {
            // 超過60s沒有使用，刪除掉
            map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
        }
        if (selectedInvoker != null) {
            // 減去總權重
            // 關於這個地方為什麼要減去總權重，是一個很容易造成迷惑的地方
            // 我的理解：每一次調用循環 每個提供者的 當前權重 都會自增自己的權重
            // 因此在選中后（只有一個被選中），再減去總權重，正好保證了所有 WeightedRoundRobin 中當前權重之和永遠等於0
            selectedWRR.sel(totalWeight);
            return selectedInvoker;
        }
        // 理論上不會走到這個地方
        // should not happen here
        return invokers.get(0);
    }

}

LeastActiveLoadBalance

最少活躍數調用算法是指在調用時判斷此時每個服務提供者此時正在處理的請求個數，選取最小的調用

dubbo 在實現該算法時的具體邏輯如下

1. 選取所有活躍數最少的提供者
2. 如果只有一個，直接返回
3. 如果權重不同，加權隨機選擇一個
4. 如果權重相同，隨機選擇一個

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // The least active value of all invokers
        // 最少活躍數量
        int leastActive = -1;
        // The number of invokers having the same least active value (leastActive)
        // 有同樣活躍值的提供者數量
        int leastCount = 0;
        // The index of invokers having the same least active value (leastActive)
        int[] leastIndexes = new int[length];
        // the weight of every invokers
        // 每一個提供者的權重
        int[] weights = new int[length];
        // The sum of the warmup weights of all the least active invokers
        // 最少活躍提供者的總權重
        int totalWeight = 0;
        // The weight of the first least active invoker
        int firstWeight = 0;
        // Every least active invoker has the same weight value?
        // 所有的最少活躍提供者是否擁有同樣的權重值
        boolean sameWeight = true;


        // Filter out all the least active invokers
        for (int i = 0; i < length; i++) {
            Invoker<T> invoker = invokers.get(i);
            // Get the active number of the invoker
            // 活躍數量
            int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
            // Get the weight of the invoker's configuration. The default value is 100.
            // 獲取權重值
            int afterWarmup = getWeight(invoker, invocation);
            // save for later use
            // 保存權重留着後面用
            weights[i] = afterWarmup;
            // If it is the first invoker or the active number of the invoker is less than the current least active number
            // 如果是第一個提供者，或者當前活躍數量比最少的少
            if (leastActive == -1 || active < leastActive) {
                // Reset the active number of the current invoker to the least active number
                // 重置最少活躍數量
                leastActive = active;
                // Reset the number of least active invokers
                // 重置最少活躍提供者的數量
                leastCount = 1;
                // Put the first least active invoker first in leastIndexes
                // 把最少活躍提供者的索引保存起來
                leastIndexes[0] = i;
                // Reset totalWeight
                // 重置總權重
                totalWeight = afterWarmup;
                // Record the weight the first least active invoker
                // 記錄第一個最少活躍提供者的權重
                firstWeight = afterWarmup;
                // Each invoke has the same weight (only one invoker here)
                // 每個最少活躍提供者是否有同樣的權重？？？
                sameWeight = true;
                // If current invoker's active value equals with leaseActive, then accumulating.
                // 如果當前活躍數量等於最少活躍數量
            } else if (active == leastActive) {
                // Record the index of the least active invoker in leastIndexes order
                // 最少活躍提供者的索引依次放入 leastIndexes
                leastIndexes[leastCount++] = i;
                // Accumulate the total weight of the least active invoker
                // 累計最少活躍提供者的總權重
                totalWeight += afterWarmup;
                // If every invoker has the same weight?
                // 如果當前權重和第一個最少活躍的權重不同，sameWeight 設為false
                if (sameWeight && afterWarmup != firstWeight) {
                    sameWeight = false;
                }
            }
        }
        // Choose an invoker from all the least active invokers
        // 最少活躍提供者只有一個，直接返回
        if (leastCount == 1) {
            // If we got exactly one invoker having the least active value, return this invoker directly.
            return invokers.get(leastIndexes[0]);
        }
        // 如擁有不同的權重，在權重的基礎上隨機選取一個，可以參考 RandomLoadBalance，有同樣的寫法
        if (!sameWeight && totalWeight > 0) {
            // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on 
            // totalWeight.
            int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
            // Return a invoker based on the random value.
            for (int i = 0; i < leastCount; i++) {
                int leastIndex = leastIndexes[i];
                offsetWeight -= weights[leastIndex];
                if (offsetWeight < 0) {
                    return invokers.get(leastIndex);
                }
            }
        }
        // 權重相同，隨機選取一個
        // If all invokers have the same weight value or totalWeight=0, return evenly.
        return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
    }

ShortestResponseLoadBalance

最短時間調用調用算法是指預估出來每個處理完請求的提供者所需時間，然後又選擇最少最短時間的提供者進行調用，整體處理邏輯和最少活躍數算法基本相似

dubbo 在實現該算法時的具體邏輯如下

1. 選取所有預估處理時間最短的提供者
2. 如果只有一個，直接返回
3. 如果權重不同，加權隨機選擇一個
4. 如果權重相同，隨機選擇一個

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // Estimated shortest response time of all invokers
        // 最少響應時間
        long shortestResponse = Long.MAX_VALUE;
        // The number of invokers having the same estimated shortest response time
        // 最少響應時間的提供者數量
        int shortestCount = 0;
        // The index of invokers having the same estimated shortest response time
        int[] shortestIndexes = new int[length];
        // the weight of every invokers
        int[] weights = new int[length];
        // The sum of the warmup weights of all the shortest response  invokers
        // 最少響應時間的提供者的總權重
        int totalWeight = 0;
        // The weight of the first shortest response invokers
        // 第一個最少響應時間的權重
        int firstWeight = 0;
        // Every shortest response invoker has the same weight value?
        // 所有的最少響應時間提供者是否擁有同樣的權重值
        boolean sameWeight = true;

        // Filter out all the shortest response invokers
        for (int i = 0; i < length; i++) {
            Invoker<T> invoker = invokers.get(i);
            RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
            // Calculate the estimated response time from the product of active connections and succeeded average elapsed time.
            //  平均響應成功時間
            long succeededAverageElapsed = rpcStatus.getSucceededAverageElapsed();
            // 活躍的連接連接數量
            int active = rpcStatus.getActive();
            // 預估響應時間
            long estimateResponse = succeededAverageElapsed * active;
            // 獲取權重值
            int afterWarmup = getWeight(invoker, invocation);
            // 保存權重留着後面用
            weights[i] = afterWarmup;
            // Same as LeastActiveLoadBalance
            // 如果預估時間小於最少的響應時間
            if (estimateResponse < shortestResponse) {
                // 重置最少響應時間
                shortestResponse = estimateResponse;
                // 最少響應時間的提供者數量設為1
                shortestCount = 1;
                // 保存提供者下標
                shortestIndexes[0] = i;
                // 重置最少響應時間的提供者的總權重
                totalWeight = afterWarmup;
                // 重置第一個最少響應時間的權重
                firstWeight = afterWarmup;
                sameWeight = true;
                // 如果當前最少響應時間等於最少響應時間
            } else if (estimateResponse == shortestResponse) {
                // 最少最少響應時間的下標依次放入 shortestIndexes
                shortestIndexes[shortestCount++] = i;
                // 累計最少響應時間的總權重
                totalWeight += afterWarmup;
                // 如果當前權重和第一個最少響應時間的權重不同，sameWeight 設為false
                if (sameWeight && i > 0
                        && afterWarmup != firstWeight) {
                    sameWeight = false;
                }
            }
        }
        // 最少最少響應時間只有一個，直接返回
        if (shortestCount == 1) {
            return invokers.get(shortestIndexes[0]);
        }
        // 如擁有不同的權重，在權重的基礎上隨機選取一個，可以參考 RandomLoadBalance，有同樣的寫法
        if (!sameWeight && totalWeight > 0) {
            int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
            for (int i = 0; i < shortestCount; i++) {
                int shortestIndex = shortestIndexes[i];
                offsetWeight -= weights[shortestIndex];
                if (offsetWeight < 0) {
                    return invokers.get(shortestIndex);
                }
            }
        }
        // 權重相同，隨機選取一個
        return invokers.get(shortestIndexes[ThreadLocalRandom.current().nextInt(shortestCount)]);
    }

ConsistentHashLoadBalance

一致性hash算法是一種廣泛應用與分佈式緩存中的算法，該算法的優勢在於新增和刪除節點后，只有少量請求發生變動，大部分請求仍舊映射到原來的節點
為了防止節點過少，造成節點分佈不均勻，一般採用虛擬節點的方式，dubbo默認的是160個虛擬節點

網上關於一致性hash算法的文章有很多，這裏就不再多贅述，以下是dubbo中的實現，需要說明的是， 一致性hash算法中權重配置不起作用

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String methodName = RpcUtils.getMethodName(invocation);
        // {group}/{interfaceName}:{version} + methoName 獲取當前消費者的唯一標示
        String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
        // using the hashcode of list to compute the hash only pay attention to the elements in the list
        int invokersHashCode = invokers.hashCode();
        // 獲取當前消費者的一致性hash選擇器
        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
        // 如果 selector 還沒初始化，或者 invokers 已經變化，重新初始化 selector
        if (selector == null || selector.identityHashCode != invokersHashCode) {
            selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, invokersHashCode));
            selector = (ConsistentHashSelector<T>) selectors.get(key);
        }
        return selector.select(invocation);
    }
    // 一致性hash選擇器
    private static final class ConsistentHashSelector<T> {

        // 存儲hash環的數據結構 節點 -> 提供者
        private final TreeMap<Long, Invoker<T>> virtualInvokers;

        // 虛擬節點數量
        private final int replicaNumber;

        // 用來標示所有提供者是唯一標示
        private final int identityHashCode;
        // 用來存儲計算hash值參數下標的數組，例如計算第一個和第三個參數 該數組為[0,2]
        private final int[] argumentIndex;

        ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
            this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
            this.identityHashCode = identityHashCode;
            URL url = invokers.get(0).getUrl();
            // 虛擬節點數量，默認 160
            this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
            // 默認只對第一個參數進行hash
            String[] index = COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, HASH_ARGUMENTS, "0"));
            argumentIndex = new int[index.length];
            for (int i = 0; i < index.length; i++) {
                argumentIndex[i] = Integer.parseInt(index[i]);
            }
            for (Invoker<T> invoker : invokers) {
                String address = invoker.getUrl().getAddress();
                // 關於這個地方為什麼要除以4，我理解的是md5後為16字節的數組，計算hash值只需要用到四個字節，所以可以用四次
                // 因此除以4，算是一個性能優化點
                for (int i = 0; i < replicaNumber / 4; i++) {
                    // md5, 獲得一個長度為16的字節數組
                    byte[] digest = md5(address + i);
                    for (int h = 0; h < 4; h++) {
                        // 如果h=0，則用第0,1,2,3四個字節進行位運算，得出一個0-2^32-1的值
                        // 如果h=1，則用第4,5,6,7四個字節進行位運算，得出一個0-2^32-1的值
                        // 如果h=2，則用第8,9,10,11四個字節進行位運算，得出一個0-2^32-1的值
                        // 如果h=3，則用第12,13,14,15四個字節進行位運算，得出一個0-2^32-1的值
                        long m = hash(digest, h);
                        virtualInvokers.put(m, invoker);
                    }
                }
            }
        }

        public Invoker<T> select(Invocation invocation) {
            String key = toKey(invocation.getArguments());
            byte[] digest = md5(key);
            return selectForKey(hash(digest, 0));
        }
        // 根據配置生成計算hash值的key
        private String toKey(Object[] args) {
            StringBuilder buf = new StringBuilder();
            for (int i : argumentIndex) {
                if (i >= 0 && i < args.length) {
                    buf.append(args[i]);
                }
            }
            return buf.toString();
        }

        private Invoker<T> selectForKey(long hash) {
            // 找到hash值在hash環上的位置
            // ceilingEntry 方法返回大於或者等於當前key的鍵值對
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
            // 如果返回為空，說明落在了hash環中2的32次方-1的最後，直接返回第一個
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            return entry.getValue();
        }
        // 得出一個0-2^32-1的值， 四個字節組成一個長度為32位的二進制数字並轉化為long值
        private long hash(byte[] digest, int number) {
            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                    | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                    | (digest[number * 4] & 0xFF))
                    & 0xFFFFFFFFL;
        }

        private byte[] md5(String value) {
            MessageDigest md5;
            try {
                md5 = MessageDigest.getInstance("MD5");
            } catch (NoSuchAlgorithmException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            md5.reset();
            byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
            md5.update(bytes);
            return md5.digest();
        }

    }

總結

以上就是dubbo負載均衡源碼的全部解析，如果還是不明白，可以看下官方文檔的解析  
http://dubbo.apache.org/zh-cn/docs/source_code_guide/loadbalance.html

dubbo的負載均衡算法總體來說並不複雜，代碼寫的也很優雅，簡潔，看起來很舒服，而且有很多細節的處理值得稱讚，例如預熱處理，輪訓算法的平滑處理等。

我們平時使用時，可以根據自己的業務場景，選擇適合自己的算法，當然，一般情況下，默認的的隨機算法就能滿足我們的日常需求，而且隨機算法的性能足夠好。

如果覺得dubbo提供的五種算法都不能滿足自己的需求，還可以通過dubbo的SPI機制很方便的擴展自己的負載均衡算法。

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

【其他文章推薦】

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

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

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

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

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

摘要：本文針對“數據牽引改進，工具固化規範”這一思路在業務團隊落地過程中的動作流程進行詳細闡述，並明確了支撐整個流程的關鍵角色定義和組織運作形式。

目的

為實現雲服務開發的過程可信，需要基於數據對各個服務產品部的可信變革動作進行數據採集、進展可視、目標牽引、能力評估，最終用數據反映目標達成。與傳統的“基於數據晾曬驅動業務團隊改進，6+1指標度量”的運作方式有本質的區別，我們是基於統一的作業工具上產生的客觀數據呈現，識別研發過程中基本的流程斷裂點和質量缺失動作，和業務團隊達成一致的目標后，把大部分改進動作固話到作業工具中自動化承載，我們稱這個思路為“數據牽引改進，工具固化規範”，也就是我們不僅告訴業務團隊哪裡有問題，同時也要基於我們的作業工具，輔助業務團隊一起改進完善。

本文針對“數據牽引改進，工具固化規範”這一思路在業務團隊落地過程中的動作流程進行詳細闡述，並明確了支撐整個流程的關鍵角色定義和組織運作形式。

數據牽引改進，是指關注軟件交付過程中各種度量數據的收集、統計、分析和反饋，通過可視化的數據客觀反映整個研發過程的狀態，以全局視角分析系統約束點，並和業務團隊達成共識，提煉出客觀有效的改進目標；工具固化規範，針對識別出來的Gap點和重點問題進行分析，制定出可以在作業工具承載的模板規範，以及需要工程師行為做出改變的能力要求，並在作業工具上對這些規範要求的落地效果進行檢查，用數據度量改進效果。最後，對改進項目進行總結分享，打造學習型組織，不斷驅動持續改進和價值交付，牽引研發團隊模式和文化的轉變。

2020年的研發過程可信圍繞CleanCode、構建、開源、E2E追溯四個領域開展，這也是公司要求的可信變革中最基本、最重要、投入產出比最大的四個點。

整體流程說明

整個運作流程，圍繞數據，按照“定義軟件工程規範->定義數據分析模型->工具實現數據度量和分析->數據運營發現實際軟件工程活動和規範的偏差->工具輔助團隊改進->工具固化軟件工程規範”這個流程進行實施，並對最終效果進行階段性總結。隨着業務團隊能力的提升以及軟件工程規範性、開發模式的改變，對最初定義的軟件工程規範，會階段性的進行完善，循環往複、持續優化，最終讓業務團隊在遵守公司要求的研發過程可信規範的前提下，實現業務成功。

1) 定義軟件工程規範：圍繞公司可信變革的目標，BU對各個服務產品部的研發模式規範和能力要求，COE制定適合BU現狀的軟件工程規範；

2) 定義數據模型：COE針對制定的軟件工程規範，提煉出核心的、有針對性、可用工具度量的數據模型，並且和各個服務產品部達成一致；

3) 工具實現數據度量和分析：根據這幾個數據模型，數據分析工具自動從數據源進行採集、匯總、計算，並把結果呈現在數據看板上；業務團隊可以打開匯總數據，根據明細數據進行動作規範自檢和改進；

4) 數據運營發現實際軟件工程活動和規範的偏差：數據治理小組在實際運營過程中，分析度量指標的數據，識別業務團隊實際的軟件工程活動和要求規範不一致的Gap點和關鍵問題；

5) 工具輔助業務團隊改進：COE針對分析出來的Gap點和關鍵問題，制定相應的改進措施，作業工具承載流程規範模板化整改，並針對業務團隊的不規範行為，制定適合各個服務產品部的公約要求，促使業務團隊人員能力提升；

6) 工具固化軟件工程規範：針對業務團隊的公約要求，在作業工具上進行check，最終作業工具承載了整個軟件工程規範要求，以及融入到作業流程中的規範要求事前檢查。

三層數據分析模型

我們採用了三層數據分析模型，由作業工具自動採集用戶研發過程行為明細數據，數據分析工具進行准實時匯總計算呈現總體目標，三層數據系統性的輔助業務團隊系統性的識別研發過程中的不規範點和能力短板，讓業務團隊“知其然，知其所以然”。這三層數據模型是層層深入，迭代完善，下層支撐上層的關係。

第一層：目標、進展、結果數據分析；和公司可信變革目標牽引對齊，結合BU實際情況，形成BU的整體可信要求，並在數據分析看板上呈現各個服務產品部要達成的過程可信目標、每日的改進進展和最終完成情況；例如，對各個服務產品部要求達成CleanCode的目標。

第二層：詞法/語法分析數據；COE針對第一層的目標牽引，分解出來的具體實施環節的度量指標，只有這些分解的指標都完成，第一層的目標才達成。這一層數據的目的主要是圍繞幫助業務團隊分析自己的能力短板在哪裡，進行有針對性的改進指；通過打開匯總數據的層層下鑽，用明細數據來分析業務團隊在DevSecOps軟件工程規範流程中關鍵動作執行的缺失點，並針對性的制定改進規範要求，牽引作業工具或者業務團隊補齊該部分缺失動作；例如，CleanCode的過程可信目標達成，可以分解成：靜態檢查配置合規率、Committer合入保障率、代碼倉Clean三個目標，只有這三個目標達成，就可以認為CleanCode總體目標達成。

第三層：語義分析數據：COE打開第二層數據，不僅要看這些關鍵動過做沒做，還要看做的效果怎麼樣，最終效果體現在業務團隊的DevSecOps軟件工程規範提升；這一層的數據分析聚焦在防止為了指標而做第二層的數據，而是看業務團隊是否在真正參考BU制定的規範牽引的目標，提升業務交付過程中的效能、可信、質量能力，以及最終產生實際的業務效果。通過打開各個團隊的明細數據分析審視業務團隊執行的關鍵動作是否符合規範，是否在合適的階段點執行，執行效果是否有效；並階段性的總結和提煉經驗，形成知識資產固化到作業工具。例如，針對第二層的靜態檢查配置合規率，可以分解為：靜態檢查配置有效性和靜態檢查執行有效性。靜態檢查配置有效性，包括：檢查靜態檢查工具配置的數量、是否符合BU的配置規範，以及是否在代碼合入主幹master時進行了配置；靜態檢查執行有效性，主要看是否每一次MR提交時都執行靜態檢查、是否發現問題在研發活動的最早階段，攔截的問題的效果怎麼樣。只有第三層的動作度量都達成后，才可以說第二層的目標是達成的。

數據治理過程流程圖

為了實現“數據牽引改進，工具固化規範”這個目標，準確、一致、准實時的數據是核心關鍵，但因為數據採集不完整、業務團隊不規範、數據呈現維度不一致等原因，數據的準確性有一個不斷提升的過程，因此需要對各個層級展示的數據進行治理。整個數據治理過程中，由“業務團隊/作業工具/治理小組/數據分析工具（阿基米德）/COE”五個角色緊密配合，而且以年/半年為目標，不斷總結經驗，循環往複、持續優化的過程。

a) COE：和公司可信變革目標牽引對齊，結合BU能力現狀，形成BU的整體可信要求；

b) COE：針對BU的業務現狀，定義出適合BU現狀的軟件工程規範要求；業務團隊：和BU發布的各個領域的軟件工程規範牽引目標達成一致；

c) COE：針對規範分解出核心的度量指標，並制定度量數據模型；

d) 研發用戶：在使用作業工具進行研發活動；作業工具：承載了BU各個服務產品部在使用過程中沉澱的行為數據；

e) 數據分析（阿基米德）：准實時接入作業工具的數據，展示各個服務產品部當前的研發能力現狀；

f) COE：和各個服務產品部達成一致，制定各個服務產品部的年度牽引目標；

g) 數據分析（阿基米德）：用數據呈現各個服務產品部的牽引目標和能力現狀，統一數據口徑；呈現月/周/天的明細數據，以及支撐Gap分析和重點問題的數據視圖；

h) COE：根據牽引目標和能力現狀，分析Gap原因和關鍵問題；治理小組：在數據運營過程中，根據數據分析團隊當前的能力現狀是否和數據呈現一致；

i) 研發用戶：可以實時登錄數據工具（阿基米德）進行查看各個層級的明細數據；

j) 治理小組：根據准實時進展數據，分析當前團隊研發過程中的實際問題，並匯總給COE；

k) COE：結合細粒度的分析數據、以及治理小組匯總出來的各個服務產品部的實際問題，制定規範和改進措施，包括作業工具的規範和研發用戶的動作行為公約；

l) 作業工具：承載作業工具上落地的規範要求；治理小組：作為接口人，承接研發工程師的行為規範公約，結合各個服務產品部實際情況來負責落地；

m) 研發用戶：按照規範要求和針對數據的自檢進行研發過程行為規範化；

n) 研發工具：對研發用戶的行為規範是否滿足要求進行自動化檢查；最終目標是讓整個軟件工程規範都固化在工具中進行承載；

o) 數據分析（阿基米德）：呈現按照規範改進后的明細數據和匯總目標；研發用戶：自助查看整改后的明細數據；

p) COE：根據數據改進的效果，以及過程中暴露的問題進行總結后形成經驗資產，並持續改進；

數據流圖

過程可信的數據在各個工具系統中採集、流轉、匯聚、分析、呈現，整個數據流圖如下：

其中，識別出6個重要的全量數據源：

a) 代碼庫數據：該數據由伏羲的服務信息樹上配置的代碼庫數據，加上阿基米德上人工配置的代碼庫，構成各個雲服務發布到生產倉的代碼全集；

b) Workitem信息流數據：當前識別vision上的需求、問題、task，加上Gitlab/Codeclub上的issue，構成可識別的Workitem數據全集；

c) SRE現網包數據：包括普羅部署、CCE、CPS、CDK各種類型部署的包數據，構成全量現網包數據；

d) 開源二進制包數據：開源中心倉數據（java、python、go、nodejs四種）語言，加上公司c/c++的數據構成全量開源二進制包數據；

e) 研發過程配置數據：阿基米德上配置的committer數據是全量的committer數據；阿基米德上識別出來的主分支是全量的主分支（邏輯“master”）數據；

f) 伏羲研發過程數據：伏羲三個庫，MongoDB的靜態檢查、門禁數據；MySQL中的測試、發布數據；MySQL中包和多個流水線的對應關係數據；一起構成了以“包”為維度的全量伏羲研發過程數據；

運作組織

數據治理運營團隊

按照過程可信在BU的落地策略，在CleanCode、構建、開源、E2E追溯四個領域設置數據治理運營團隊，由 “數據分析工具（阿基米德）—COE—各個服務產品部接口人組成的治理小組”三個角色組成，以“指標度量為牽引，數據的客觀呈現為落地方式，業務的價值反饋為最終目的”的原則來落地數據治理工作。

COE的職責：

1） 和公司可信變革目標牽引對齊，結合BU能力現狀，形成BU的整體可信要求；定義出適合BU現狀的軟件工程規範要求；針對規範分解出核心的度量指標，並制定度量數據模型；

2） 利用作業工具已經產生的數據，和治理小組一起分析識別數據質量的問題，按照三層數據分析模型，層層打開，識別業務團隊能力Gap點。

3） 分析典型問題，識別作業流的斷裂點進行補齊，和業務團隊的不規範動作，制定規範和公約要求，逐步改善數據質量。

4） 事後歸納總結，識別出流程缺失，組織缺失，責任缺失等機制行問題，並固化到作業工具中。

治理小組：

1） 結合各個服務產品部的實際情況，承接COE的數據治理規範在各個服務產品部的落地；

2） 識別數據治理動作在各個服務產品部落地過程中的實際問題，和COE一起分析，提出系統性的解決思路，最終固化到作業工具中。

3） 跟蹤過程可信在業務團隊落地的過程中的進展，為業務團隊最終達成可信變革目標負責，為改進過程產生實際的業務價值負責；

數據分析工具（阿基米德）：

1） 確保接入的數據準確、實時、一致，用數據實時反映BU各個服務產品部的能力現狀，為COE和治理小組的數據運營提供數據支撐；

2） 系統性的落地COE的方案設計，實現整個BU統一標準的數據看板，能夠清晰的通過數據識別出來業務團隊的能力Gap，牽引業務團隊達成整體改進目標；

3） 按照三層數據模型進行數據展示，層層下鑽，讓業務團隊“知其然，知其所以然”，牽引業務團隊中的每一個人都能自己進行改進；

4） 通過數據分析，識別DevSecOps軟件工程規範在BU的業務團隊落地過程中的重點問題，以及該問題背後的流程、制度缺失，促使最終規範固化在作業工具中。

例會設置

“數據驅動DevSecOps能力提升例會”為研發領域數據治理相關問題的求助和裁決例會。

會議分為三個階段：

1） 第一階段，例行議題，形式類似於“體檢報告”，用數據反映業務團隊的現狀和問題；

2） 第二階段，申報議題，形式類似於“專家會診”，討論某一個具體數據治理過程中的問題和Top困難求助；

3） 第三階段，靈活安排議題，形式類似於“問題總結”，針對某一類的具體問題，進行集中討論和歸納總結定義，形成BU的規範流程和章程總結。

主數據承載系統

主數據是指具有高業務價值的、可以在企業內跨越多個業務部門被重複使用的數據，是單一、準確、權威的數據來源。和業務型數據、分析型數據相比，主數據主要有以下幾個特徵：

1） 特徵一致性：也就是能否保證主數據的關鍵特徵在不同應用、不同系統中的高度一致，直接關係了數據治理成敗；

2） 識別唯一性：在一個系統、一個平台，甚至一個企業範圍內，同一主數據實體要求具有唯一的數據標識，即數據編碼；

3） 長期有效性：貫穿該業務對象的整個生命周期甚至更長，當該主數據失去其效果時，系統採取的措施通常為軟刪除，而不是物理刪除；

4） 業務穩定性：在業務過程中其識別信息和關鍵特徵會被業務過程中產生的數據繼承、引用和複製。除非該主數據本身的特徵發生變化，否則該主數據不會隨着業務的過程中被其他系統修改。

主數據源識別原則：

a) 如果有多個數據源構成同類型數據的主數據，兩種處理策略：

1）選取一個源系統逐步收編其他源系統的數據，變成唯一主數據源

2）如果1）不能實現，由阿基米德系統進行封裝后屏蔽多個數據源系統，該類型數據的唯一數據源變成阿基米德，待後續1）實現后，阿基米德該類型主數據源失效。

3）當數據在多個作業系統中進行流轉時，判斷是否作為主數據源的標準是：數據在該系統有實際的業務動作產生，而不是只承載數據的流轉。

b) 如果確定為唯一數據源，其他消費該類型數據的系統不能和數據源產生衝突。

所有數據僅能在數據源產生，其它系統只能讀取不能修改。下游發現的數據源質量問題，應當在數據源頭進行修正。

c) 主數據使用方不得改變原始數據，但可以進行擴展。

數據消費方不得對獲取的數據進行增、刪、改，但可以在數據的基礎上進行屬性擴展。

d) 在滿足信息安全的前提下充分共享，不得拒絕合理的數據共享需求。

數據如果不流轉，不僅不會產生業務價值，還增加存儲成本；只有不斷流轉，對業務團隊產生實際價值時，還能得到使用效果的反饋，促進數據價值的進一步提升。

原則為：核心資產安全優先，非關鍵資產效率優先。

一類主數據源

二類主數據源

點擊關注，第一時間了解華為雲新鮮技術~

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

【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

※別再煩惱如何寫文案,掌握八大原則!