碼奴蔣祥機

異步"偽線程"重構《手搓》線程池,支持任務清退

網友@舟翅桐和@0611163建議Task清退機制多次嘗試不能很好實現清退使用異步"偽線程"推倒重構《手搓》線程池重見光明

一、為什么需要Task清退

大家有沒有點到過這樣的按鈕

點完之后轉圈圈,頁面卡死

多希望盡快彈出一個是否取消的按鈕

如果頁面的關閉按鈕還能用,會毫不猶豫的去點

可想而知長耗時任務如果沒有取消功能是多差的用戶體驗

二、再說說Task如何清退

Task可以通過CancellationToken實現清退

大部分IO操作都支持CancellationToken

比如EFCore、Dapper、HttpClient等,都支持CancellationToken

異步方法一般都建議包含CancellationToken參數

如果把同步方法比作碼奴心愛的玩具

那異步方法就好比是能上天的高級玩具風箏

CancellationToken就是那根風箏線

有了CancellationToken,我們的異步方法可以做到收放自如

即使"開弓"也能有"回頭箭"

1. 通過ThrowIfCancellationRequested清退的Case

本Case計算1000累加,每次計算耗時為當前值的毫秒數

預估耗時500秒

如果用戶取消通過ThrowIfCancellationRequested觸發異常終止任務

通過CancellationTokenSource構造CancellationToken

可以通過CancelAfter設置超時時間,時間過了自動取消

還可以手動調用Cancel方法取消

本Case設置10秒后超時

發起異步1秒后調用Cancel

結果觸發異常

實際耗時1秒

避免了500秒的等待

是一個非常成功的清退

int result = 0;
var tokenSource = new CancellationTokenSource();
tokenSource.CancelAfter(TimeSpan.FromSeconds(10));
var sw = Stopwatch.StartNew();
var task = CountAsynWithThrowIfCancellationRequested(1000, tokenSource.Token);
await Task.Delay(1000, CancellationToken.None);
tokenSource.Cancel();
try
{
    result = await task;
}
catch (Exception ex)
{
    _output.WriteLine(ex.ToString());
}
sw.Stop();
_output.WriteLine($"Result: {result} Elapsed:{sw.Elapsed.TotalMilliseconds}");

private static async Task<int> CountAsynWithThrowIfCancellationRequested(int num, CancellationToken token)
{
    var count = 0;
    for (int i = 0; i < num; i++)
    {
        await Task.Delay(i, CancellationToken.None);
        token.ThrowIfCancellationRequested();
        count += i;
    }
    return count;
}

// System.OperationCanceledException: The operation was canceled.
//    at System.Threading.CancellationToken.ThrowOperationCanceledException()
//    at System.Threading.CancellationToken.ThrowIfCancellationRequested()
//    at TaskTests.Tasks.CancellationTokenTests.CountAsynWithThrowIfCancellationRequested(Int32 num, CancellationToken token) in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\CancellationTokenTests.cs:line 51
//    at TaskTests.Tasks.CancellationTokenTests.ThrowIfCancellationRequested() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\CancellationTokenTests.cs:line 22
// Result: 0 Elapsed:1028.8541

2. 通過IsCancellationRequested清退的Case

前面Case有個問題

雖然我們無法忍受500秒拿到最終結果

但是已經等待了1秒了,能不能把這1秒的結果先給我,也算沒白等

通過IsCancellationRequested可以實現

還是前面那個Case

這次還不用catch了

實際耗時1秒,拿到中間結果703

var tokenSource = new CancellationTokenSource();
tokenSource.CancelAfter(TimeSpan.FromSeconds(10));
var sw = Stopwatch.StartNew();
var task = CountAsynWithIsCancellationRequested(1000, tokenSource.Token);
await Task.Delay(1000, CancellationToken.None);
tokenSource.Cancel();
var result = await task;
sw.Stop();
_output.WriteLine($"Result: {result} Elapsed:{sw.Elapsed.TotalMilliseconds}");

private static async Task<int> CountAsynWithIsCancellationRequested(int num, CancellationToken token)
{
    var count = 0;
    for (int i = 0; i < num; i++) 
    {            
        await Task.Delay(i, CancellationToken.None);
        if (token.IsCancellationRequested)
            break;
        count += i;
    }
    return count;
}

// Result: 703 Elapsed:1056.0416

3. 通過CreateLinkedTokenSource實現復雜的清退規則

如下復雜業務邏輯

執行A、B兩個邏輯的結果再調用C邏輯

總共耗時不能超過1秒

其中A、B邏輯不能超過800毫秒,C邏輯不能超過600毫秒

為了更好實現需求,A和B并行節約時間

用CreateLinkedTokenSource實現C操作要同時滿足總耗時不超過1秒,C本身不超過600毫秒

綜上CancellationToken作用很大,可以設置超時、可以手動觸發還可以支持多條件組合

var tokenSource = new CancellationTokenSource();
tokenSource.CancelAfter(TimeSpan.FromSeconds(1));

var tokenSource1 = new CancellationTokenSource();
tokenSource1.CancelAfter(TimeSpan.FromSeconds(800));
var token1 = tokenSource1.Token;
var taskA = A(500, token1);
var taskB = B(400, token1);
var a = await taskA;
var b = await taskB;

var cancellationToken2 = new CancellationTokenSource();
cancellationToken2.CancelAfter(TimeSpan.FromSeconds(600));
var linked = CancellationTokenSource.CreateLinkedTokenSource(tokenSource.Token, cancellationToken2.Token);
var taskC = C(400, a, b, linked.Token);
var c = await taskC;
Assert.Equal(3, c);

private static async Task<int> A(int arg, CancellationToken token)
{
    await Task.Delay(arg, token);
    return 1;
}
private static async Task<int> B(int arg, CancellationToken token)
{
    await Task.Delay(arg, token);
    return 2;
}
private static async Task<int> C(int arg, int a, int b, CancellationToken token)
{
    await Task.Delay(arg, token);
    return a + b;
}

三、《手搓》線程池可清退任務

1. 單個異步任務清退Case

通過processor.AddTask添加異步任務并啟動線程池

通過tokenSource.Cancel()取消

任務最終并未執行

異步方法最好添加CancellationToken參數以便更精細化處理

特別是邏輯比較復雜的方法和循環處理,減少不必要的等待和無效的CPU計算

var options = new ReduceOptions { ConcurrencyLevel = 1, AutoStart = false };
var processor = new Processor();
var pool = options.CreateJob(processor);
var tokenSource = new CancellationTokenSource();
var token = tokenSource.Token;
var state = processor.AddTask((t) => HelloAsync("張三", t), token);
pool.Start();
tokenSource.Cancel();
await Task.Delay(1000);
Assert.True(state.IsCancel);

async Task HelloAsync(string name, CancellationToken token = default)
{
    await Task.Delay(10, token);
    _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} HelloAsync {name},{DateTime.Now:HH:mm:ss.fff}");
}

2. 單個同步任務清退Case

通過processor.Add添加同步任務并啟動線程池

通過tokenSource.Cancel()取消

任務最終并未執行

var options = new ReduceOptions { ConcurrencyLevel = 1, AutoStart = false };
var processor = new Processor();
var pool = options.CreateJob(processor);
var tokenSource = new CancellationTokenSource();
var state = processor.Add(() => Hello("張三"), tokenSource.Token);
pool.Start();
tokenSource.Cancel();
await Task.Delay(1000);
Assert.True(state.IsCancel);

void Hello(string name, int time = 10)
{
    Thread.Sleep(time);
    _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Hello {name},{DateTime.Now:HH:mm:ss.fff}");
}

四、《手搓》線程池可清退線程

1. 堵塞線程池的Case

ConcurrencyLevel設置為1

這次先添加10個正常的任務

再通過Token添加bug,耗時是其他任務的100倍,并設置了該任務1秒超時

后面又添加了90個任務

從執行結果可以看到,執行前10個任務后確實阻塞了線程池1秒

1秒后線程池恢復繼續執行剩下的90個任務

有一個細節,Bug那個任務也執行了,插在第48個任務之后

如果方法已經執行很可能無法真實的取消(除非增加token參數來控制)

但是可以把當前“線程”回收,避免由此可能導致的線程池堵塞

上面的線程筆者特意加了引號,這里說的“線程”實際是一個“線程配額”,來自系統線程池

回收也是一個配額,原方法一旦開始運行只能等他自行結束

技術上停止線程也是可以實現的,但這可能導致不可預期的后果,強烈反對強行終止線程

而手搓線程要做的是找系統線程池再要一個“配額”

特別提醒不要以為不會堵塞《手搓》線程池就可以隨便加超時任務

最終消耗的都是系統線程池的資源

當系統線程池耗完,整個程序就不好了,當然《手搓》線程池也會成為無源之水，無本之木

var options = new ReduceOptions { ConcurrencyLevel = 1 };
var processor = new Processor();
var pool = options.CreateJob(processor);
for (int i = 0; i < 10; i++)
{
    var user = "User" + i;
    processor.Add(() => Hello(user, 20));
}
var bugToken = new CancellationTokenSource();
bugToken.CancelAfter(TimeSpan.FromMilliseconds(1000));
processor.Add(() => Hello("Bug", 2000), bugToken.Token);
for (int i = 10; i < 100; i++)
{
    var user = "User" + i;
    processor.Add(() => Hello(user, 20));
}
await Task.Delay(5000);

void Hello(string name, int time = 10)
{
    Thread.Sleep(time);
    _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Hello {name},{DateTime.Now:HH:mm:ss.fff}");
}

// Thread11 Hello User0,00:50:43.376
// Thread11 Hello User1,00:50:43.408
// Thread11 Hello User2,00:50:43.440
// Thread11 Hello User3,00:50:43.472
// Thread11 Hello User4,00:50:43.504
// Thread11 Hello User5,00:50:43.536
// Thread11 Hello User6,00:50:43.568
// Thread11 Hello User7,00:50:43.600
// Thread11 Hello User8,00:50:43.632
// Thread11 Hello User9,00:50:43.664
// Thread31 Hello User10,00:50:44.447
// Thread31 Hello User11,00:50:44.479
// Thread31 Hello User12,00:50:44.511
// Thread31 Hello User13,00:50:44.543
// Thread31 Hello User14,00:50:44.575
// Thread31 Hello User15,00:50:44.607
// Thread31 Hello User16,00:50:44.639
// Thread31 Hello User17,00:50:44.671
// Thread31 Hello User18,00:50:44.703
// Thread31 Hello User19,00:50:44.735
// Thread31 Hello User20,00:50:44.767
// Thread31 Hello User21,00:50:44.799
// Thread31 Hello User22,00:50:44.831
// Thread31 Hello User23,00:50:44.863
// Thread31 Hello User24,00:50:44.895
// Thread31 Hello User25,00:50:44.927
// Thread31 Hello User26,00:50:44.959
// Thread31 Hello User27,00:50:44.990
// Thread31 Hello User28,00:50:45.022
// Thread31 Hello User29,00:50:45.053
// Thread31 Hello User30,00:50:45.084
// Thread31 Hello User31,00:50:45.116
// Thread31 Hello User32,00:50:45.148
// Thread31 Hello User33,00:50:45.180
// Thread31 Hello User34,00:50:45.212
// Thread31 Hello User35,00:50:45.244
// Thread31 Hello User36,00:50:45.276
// Thread31 Hello User37,00:50:45.308
// Thread31 Hello User38,00:50:45.340
// Thread31 Hello User39,00:50:45.372
// Thread31 Hello User40,00:50:45.404
// Thread31 Hello User41,00:50:45.436
// Thread31 Hello User42,00:50:45.468
// Thread31 Hello User43,00:50:45.500
// Thread31 Hello User44,00:50:45.532
// Thread31 Hello User45,00:50:45.564
// Thread31 Hello User46,00:50:45.596
// Thread31 Hello User47,00:50:45.628
// Thread31 Hello User48,00:50:45.660
// Thread11 Hello Bug,00:50:45.675
// Thread31 Hello User49,00:50:45.691
// Thread32 Hello User50,00:50:45.723
// Thread32 Hello User51,00:50:45.755
// Thread32 Hello User52,00:50:45.786
// Thread32 Hello User53,00:50:45.817
// Thread32 Hello User54,00:50:45.849
// Thread32 Hello User55,00:50:45.881
// Thread32 Hello User56,00:50:45.913
// Thread32 Hello User57,00:50:45.945
// Thread32 Hello User58,00:50:45.977
// Thread32 Hello User59,00:50:46.009
// Thread32 Hello User60,00:50:46.041
// Thread32 Hello User61,00:50:46.073
// Thread32 Hello User62,00:50:46.105
// Thread32 Hello User63,00:50:46.137
// Thread32 Hello User64,00:50:46.169
// Thread32 Hello User65,00:50:46.201
// Thread32 Hello User66,00:50:46.233
// Thread32 Hello User67,00:50:46.265
// Thread32 Hello User68,00:50:46.297
// Thread32 Hello User69,00:50:46.329
// Thread32 Hello User70,00:50:46.361
// Thread32 Hello User71,00:50:46.393
// Thread32 Hello User72,00:50:46.425
// Thread32 Hello User73,00:50:46.457
// Thread32 Hello User74,00:50:46.489
// Thread32 Hello User75,00:50:46.521
// Thread32 Hello User76,00:50:46.552
// Thread32 Hello User77,00:50:46.584
// Thread32 Hello User78,00:50:46.616
// Thread32 Hello User79,00:50:46.648
// Thread32 Hello User80,00:50:46.680
// Thread32 Hello User81,00:50:46.712
// Thread32 Hello User82,00:50:46.744
// Thread32 Hello User83,00:50:46.776
// Thread32 Hello User84,00:50:46.808
// Thread32 Hello User85,00:50:46.840
// Thread32 Hello User86,00:50:46.872
// Thread32 Hello User87,00:50:46.904
// Thread32 Hello User88,00:50:46.936
// Thread32 Hello User89,00:50:46.967
// Thread32 Hello User90,00:50:46.999
// Thread32 Hello User91,00:50:47.031
// Thread32 Hello User92,00:50:47.063
// Thread32 Hello User93,00:50:47.095
// Thread32 Hello User94,00:50:47.127
// Thread32 Hello User95,00:50:47.159
// Thread32 Hello User96,00:50:47.191
// Thread32 Hello User97,00:50:47.222
// Thread32 Hello User98,00:50:47.254
// Thread32 Hello User99,00:50:47.286

2. 沒有token參數的任務堵塞線程池怎么辦

這次增加了參數ItemLife,設置為1秒

依然是添加10個任務,插入一個Bug,再添加90個任務

這次Bug沒有設置token

效果跟上次差不多,線程池阻塞1秒

Bug插入40之后

也就是說ItemLife提供了全局保護

再配合前面的token,可以有效提供線程池的可用性

必須強調一下,為了測試博文中設置的線程池都很小

這屬于邊界測試,實際項目建議線程池盡量設大一點,不會打掛上游就行

如果線上高并發項目也像本測試這樣,線程池阻塞1秒是完全無法接受的

var options = new ReduceOptions { ConcurrencyLevel = 1, ItemLife = TimeSpan.FromSeconds(1) };
var processor = new Processor();
var pool = options.CreateJob(processor);
for (int i = 0; i < 10; i++)
{
    var user = "User" + i;
    processor.Add(() => Hello(user, 20));
}
processor.Add(() => Hello("Bug", 2000));
for (int i = 10; i < 100; i++)
{
    var user = "User" + i;
    processor.Add(() => Hello(user, 20));
}
await Task.Delay(5000);

// Thread11 Hello User0,02:41:30.413
// Thread11 Hello User1,02:41:30.445
// Thread11 Hello User2,02:41:30.477
// Thread11 Hello User3,02:41:30.509
// Thread11 Hello User4,02:41:30.540
// Thread11 Hello User5,02:41:30.571
// Thread11 Hello User6,02:41:30.601
// Thread11 Hello User7,02:41:30.632
// Thread11 Hello User8,02:41:30.664
// Thread11 Hello User9,02:41:30.696
// Thread31 Hello User10,02:41:31.746
// Thread31 Hello User11,02:41:31.778
// Thread31 Hello User12,02:41:31.810
// Thread31 Hello User13,02:41:31.842
// Thread31 Hello User14,02:41:31.874
// Thread31 Hello User15,02:41:31.906
// Thread31 Hello User16,02:41:31.938
// Thread31 Hello User17,02:41:31.970
// Thread31 Hello User18,02:41:32.002
// Thread31 Hello User19,02:41:32.034
// Thread31 Hello User20,02:41:32.066
// Thread31 Hello User21,02:41:32.098
// Thread31 Hello User22,02:41:32.130
// Thread31 Hello User23,02:41:32.162
// Thread31 Hello User24,02:41:32.194
// Thread31 Hello User25,02:41:32.226
// Thread31 Hello User26,02:41:32.258
// Thread31 Hello User27,02:41:32.289
// Thread31 Hello User28,02:41:32.321
// Thread31 Hello User29,02:41:32.353
// Thread31 Hello User30,02:41:32.385
// Thread31 Hello User31,02:41:32.417
// Thread31 Hello User32,02:41:32.449
// Thread31 Hello User33,02:41:32.481
// Thread31 Hello User34,02:41:32.513
// Thread31 Hello User35,02:41:32.545
// Thread31 Hello User36,02:41:32.577
// Thread31 Hello User37,02:41:32.609
// Thread31 Hello User38,02:41:32.641
// Thread31 Hello User39,02:41:32.673
// Thread31 Hello User40,02:41:32.705
// Thread11 Hello Bug,02:41:32.705
// Thread31 Hello User41,02:41:32.737
// Thread8 Hello User42,02:41:32.769
// Thread8 Hello User43,02:41:32.801
// Thread8 Hello User44,02:41:32.833
// Thread8 Hello User45,02:41:32.865
// Thread8 Hello User46,02:41:32.897
// Thread8 Hello User47,02:41:32.929
// Thread8 Hello User48,02:41:32.961
// Thread8 Hello User49,02:41:32.993
// Thread8 Hello User50,02:41:33.025
// Thread8 Hello User51,02:41:33.057
// Thread8 Hello User52,02:41:33.089
// Thread8 Hello User53,02:41:33.121
// Thread8 Hello User54,02:41:33.153
// Thread8 Hello User55,02:41:33.185
// Thread8 Hello User56,02:41:33.217
// Thread8 Hello User57,02:41:33.249
// Thread8 Hello User58,02:41:33.281
// Thread8 Hello User59,02:41:33.313
// Thread8 Hello User60,02:41:33.345
// Thread8 Hello User61,02:41:33.377
// Thread8 Hello User62,02:41:33.409
// Thread8 Hello User63,02:41:33.441
// Thread8 Hello User64,02:41:33.473
// Thread8 Hello User65,02:41:33.505
// Thread8 Hello User66,02:41:33.537
// Thread8 Hello User67,02:41:33.568
// Thread8 Hello User68,02:41:33.600
// Thread8 Hello User69,02:41:33.632
// Thread8 Hello User70,02:41:33.664
// Thread8 Hello User71,02:41:33.696
// Thread8 Hello User72,02:41:33.728
// Thread8 Hello User73,02:41:33.759
// Thread8 Hello User74,02:41:33.791
// Thread8 Hello User75,02:41:33.823
// Thread8 Hello User76,02:41:33.855
// Thread8 Hello User77,02:41:33.887
// Thread8 Hello User78,02:41:33.919
// Thread8 Hello User79,02:41:33.951
// Thread8 Hello User80,02:41:33.983
// Thread8 Hello User81,02:41:34.015
// Thread8 Hello User82,02:41:34.047
// Thread8 Hello User83,02:41:34.079
// Thread8 Hello User84,02:41:34.111
// Thread8 Hello User85,02:41:34.143
// Thread8 Hello User86,02:41:34.174
// Thread8 Hello User87,02:41:34.206
// Thread8 Hello User88,02:41:34.238
// Thread8 Hello User89,02:41:34.270
// Thread8 Hello User90,02:41:34.302
// Thread8 Hello User91,02:41:34.333
// Thread8 Hello User92,02:41:34.365
// Thread8 Hello User93,02:41:34.397
// Thread8 Hello User94,02:41:34.428
// Thread8 Hello User95,02:41:34.460
// Thread8 Hello User96,02:41:34.491
// Thread8 Hello User97,02:41:34.523
// Thread8 Hello User98,02:41:34.555
// Thread8 Hello User99,02:41:34.587

五、追蹤《手搓》線程池任務狀態

1. 追蹤同步任務狀態的Case

添加Action任務會返回一個state

任務尚未執行IsSuccess為false

任務執行成功IsSuccess為true

另外state還有屬性IsCancel,為true時表示任務已經取消

Exception屬性表示任務執行過程中觸發的異常

《手搓》線程池以上特性是不是比系統線程池要方便不少

另外請大家放心,任務狀態信息通過回調賦值,對性能幾乎沒有影響

var options = new ReduceOptions { ConcurrencyLevel = 1, AutoStart = false };
var processor = new Processor();
var pool = options.CreateJob(processor);
var state = processor.Add(() => Hello("張三"));
Assert.False(state.IsSuccess);
pool.Start();
await Task.Delay(1000);
Assert.True(state.IsSuccess);

/// <summary>
/// 任務狀態
/// </summary>
public interface IJobState
{
    /// <summary>
    /// 是否執行成功
    /// </summary>
    bool IsSuccess { get; }
    /// <summary>
    /// 是否執行失敗
    /// </summary>
    bool IsFail { get; }
    /// <summary>
    /// 是否取消
    /// </summary>
    bool IsCancel { get; }
    /// <summary>
    /// 異常
    /// </summary>
    public Exception Exception { get; }
}

2. 只執行不追蹤任務狀態可以嗎

當然可以

《手搓》線程池提供了一個簡單的處理器ActionProcessor,專治性能強迫癥患者

ActionProcessor.Instance是默認實例,只有執行邏輯,多個線程池可以共用

ActionProcessor和pool的Add方法是void類型

ActionProcessor只執行不回調任務狀態

ActionProcessor有個缺點(也可能是優點),只支持同步任務

另外ActionProcessor不支持token設置單個任務取消

ItemLife全局保護還是支持的

var options = new ReduceOptions { ConcurrencyLevel = 1 };
var pool = options.CreateJob(ActionProcessor.Instance);
pool.Add(() => Hello("張三"));
pool.Add(() => Hello("李四"));

// Thread11 Hello 張三,03:09:42.222
// Thread11 Hello 李四,03:09:42.241

3. 追蹤異步任務狀態的Case

添加異步任務也會返回一個state

與同步任務一樣

var options = new ReduceOptions { ConcurrencyLevel = 1, AutoStart = false };
var processor = new Processor();
var pool = options.CreateJob(processor);
var state = processor.AddTask(() => HelloAsync("張三"));
Assert.False(state.IsSuccess);
pool.Start();
await Task.Delay(1000);
Assert.True(state.IsSuccess);

六、獲取《手搓》線程池任務執行結果

1. 獲取同步任務執行結果的Case

添加Func任務會返回一個result

result類型繼承前面的IJobState,并多一個Result屬性

var options = new ReduceOptions { ConcurrencyLevel = 1, AutoStart = false };
var processor = new Processor();
var pool = options.CreateJob(processor);
var result = processor.Add(() => Count(3));
Assert.False(result.IsSuccess);
pool.Start();
await Task.Delay(1000);
Assert.True(result.IsSuccess);
var count = result.Result;
Assert.Equal(6, count);

static int Count(int num)
{
    int result = 0;
    for (int i = 1; i <= num; i++)
        result += i;
    return result;
}

/// <summary>
/// 任務執行結果
/// </summary>
/// <typeparam name="TResult"></typeparam>
public interface IJobResult<out TResult>
    : IJobState
{
    /// <summary>
    /// 結果
    /// </summary>
    TResult Result { get; }
}

2. 獲取異步任務執行結果的Case

添加Func異步任務也會返回一個result

當然這個result不能代替Task,不能通過await等到結果完成直接使用

這些效果還是要靠手搓TaskFactory來實現

手搓TaskFactory是基于手搓線程池實現的,這次手搓線程池大范圍重構

手搓TaskFactory也是重構了,抽空筆者再補一篇手搓TaskFactory重構的文章

var options = new ReduceOptions { ConcurrencyLevel = 1, AutoStart = false };
var processor = new Processor();
var pool = options.CreateJob(processor);
var tokenSource = new CancellationTokenSource();
tokenSource.CancelAfter(TimeSpan.FromSeconds(1));
var result = processor.AddTask((t) => CountAsync(3, t), tokenSource.Token);
Assert.False(result.IsSuccess);
pool.Start();
await Task.Delay(1000);
Assert.True(result.IsSuccess);
var count = result.Result;
Assert.Equal(6, count);

static async Task<int> CountAsync(int num, CancellationToken token = default)
{
    int result = 0;
    for (int i = 1; i <= num; i++)
    {
        await Task.Delay(1, token);
        result += i;
    }    
    return result;
}

七、揭秘手搓線程池重構

1. 重構后的手搓線程池

手搓線程池還是由"主線程"和真實線程池構成

區別在于"主線程"的職責的發生了變化

當然"線程"也發生了很大的變化,由真線程變為"偽線程"

2. "主線程"的變化

"主線程"不再執行任務,考慮到任務可能阻塞線程

如果先阻塞了主線程,繼而其他線程都執行完回收后,再突發任務,會導致"餓死"線程池的不良后果

就是任務堆積,線程池沒滿但就是沒線程在執行

主線程只做3件事

其一就是檢查有無線程被阻塞,對被阻塞的線程進行回收

其二是否有任務需要執行,如果有任務就激活一個線程

其三就是休眠一段時間,通過ReduceTime配置,默認50毫秒

3. 真線程變為"偽線程"

由于需要支持異步,如果用真線程await異步操作,那就是浪費一個線程

所以重構為從系統線程池"申請"線程"配額",await的時候線程還給系統,系統可以另行安排

await完成線程再次激活,當然不見得還是前面那個線程,所以變成了"偽線程",也可以說是一個線程"配額"

4. 線程增加狀態

增加了LastTime屬性,用于監控線程是否被堵塞

增加了LastItem屬性,用于監控當前執行任務狀態是否正常(是否被取消)

好了,就介紹到這里,更多信息請查看源碼庫
源碼托管地址: https://github.com/donetsoftwork/HandCore.net ，歡迎大家直接查看源碼。
gitee同步更新:https://gitee.com/donetsoftwork/HandCore.net

如果大家喜歡請動動您發財的小手手幫忙點一下Star,謝謝！！！

posted on 2025-11-05 09:32 xiangji 閱讀(78) 評論(0) 收藏舉報

刷新頁面返回頂部