一、先回顧一下以前《手搓》線程池Case

• ConcurrencyLevel設置為10
• 并發實現了完美的指數遞進關系
• 當時內心還是得到了很大的滿足
• 第一批01:58:55.832是1個并發
• 第二批01:58:55.944是2個并發
• 第三批01:58:56.054是4個并發
• 第四批01:58:56.165是8個并發
• 01:58:56.276及以后一段時間達到并發上限10個
• 參考筆者博文《手搓》線程池

var options = new ReduceOptions { ConcurrencyLevel = 10 };
var scheduler = new ConcurrentTaskScheduler(options);
var factory = new TaskFactory(scheduler);
var jobService = new ConcurrentJobService(scheduler, options);
jobService.Start();
Start(factory);

private void Start(TaskFactory factory)
{
    for (int i = 1; i < 10; i++)
    {
        for (int j = 1; j < 10; j++)
        {
            int a = i, b = j;
            factory.StartNew(() => Multiply(a, b));
        }
    }
}
public int Multiply(int a, int b)
{
    var result = a * b;
    _output.WriteLine($"{a} x {b} = {result},{DateTime.Now:HH:mm:ss.fff}");
    Thread.Sleep(100);
    return result;
}

// 1 x 1 = 1,01:58:55.832
// 1 x 2 = 2,01:58:55.944
// 1 x 3 = 3,01:58:55.944
// 1 x 4 = 4,01:58:56.054
// 1 x 5 = 5,01:58:56.054
// 1 x 6 = 6,01:58:56.054
// 1 x 7 = 7,01:58:56.054
// 1 x 8 = 8,01:58:56.165
// 1 x 9 = 9,01:58:56.165
// 2 x 1 = 2,01:58:56.165
// 2 x 3 = 6,01:58:56.165
// 2 x 2 = 4,01:58:56.165
// 2 x 4 = 8,01:58:56.165
// 2 x 5 = 10,01:58:56.165
// 2 x 6 = 12,01:58:56.165
// 2 x 8 = 16,01:58:56.276
// 2 x 9 = 18,01:58:56.276
// 3 x 2 = 6,01:58:56.276
// 2 x 7 = 14,01:58:56.276
// 3 x 3 = 9,01:58:56.276
// 3 x 1 = 3,01:58:56.276
// 3 x 4 = 12,01:58:56.276
// 3 x 5 = 15,01:58:56.276
// 3 x 6 = 18,01:58:56.277
// 3 x 7 = 21,01:58:56.277
// 4 x 1 = 4,01:58:56.388
// 3 x 8 = 24,01:58:56.388
// 3 x 9 = 27,01:58:56.388
// 4 x 2 = 8,01:58:56.388
// 4 x 6 = 24,01:58:56.388
// 4 x 3 = 12,01:58:56.388
// 4 x 5 = 20,01:58:56.388
// 4 x 8 = 32,01:58:56.388
// 4 x 4 = 16,01:58:56.388
// 4 x 7 = 28,01:58:56.388
// 5 x 8 = 40,01:58:56.500
// 5 x 4 = 20,01:58:56.500
// 5 x 9 = 45,01:58:56.500
// 5 x 2 = 10,01:58:56.500
// 5 x 7 = 35,01:58:56.500
// 5 x 6 = 30,01:58:56.500
// 4 x 9 = 36,01:58:56.500
// 5 x 5 = 25,01:58:56.500
// 5 x 1 = 5,01:58:56.500
// 5 x 3 = 15,01:58:56.500
// 6 x 5 = 30,01:58:56.612
// 6 x 2 = 12,01:58:56.612
// 6 x 1 = 6,01:58:56.612
// 6 x 3 = 18,01:58:56.612
// 6 x 6 = 36,01:58:56.612
// 6 x 7 = 42,01:58:56.612
// 6 x 4 = 24,01:58:56.612
// 6 x 8 = 48,01:58:56.612
// 6 x 9 = 54,01:58:56.612
// 7 x 1 = 7,01:58:56.612
// 7 x 2 = 14,01:58:56.724
// 8 x 1 = 8,01:58:56.724
// 7 x 4 = 28,01:58:56.724
// 7 x 7 = 49,01:58:56.724
// 7 x 5 = 35,01:58:56.724
// 8 x 2 = 16,01:58:56.724
// 7 x 8 = 56,01:58:56.724
// 7 x 6 = 42,01:58:56.724
// 7 x 9 = 63,01:58:56.724
// 7 x 3 = 21,01:58:56.724
// 8 x 5 = 40,01:58:56.836
// 8 x 8 = 64,01:58:56.836
// 9 x 1 = 9,01:58:56.836
// 9 x 2 = 18,01:58:56.836
// 8 x 7 = 56,01:58:56.836
// 8 x 9 = 72,01:58:56.836
// 8 x 3 = 24,01:58:56.836
// 8 x 6 = 48,01:58:56.836
// 9 x 3 = 27,01:58:56.836
// 8 x 4 = 32,01:58:56.836
// 9 x 5 = 45,01:58:56.948
// 9 x 7 = 63,01:58:56.948
// 9 x 4 = 36,01:58:56.948
// 9 x 6 = 54,01:58:56.948
// 9 x 9 = 81,01:58:56.948
// 9 x 8 = 72,01:58:56.948

二、短暫的快樂被異步并發測試給打破了

• ConcurrencyLevel設置為4
• 一開始就是4個清晰可見的并發
• 一發入魂,開局即是高潮
• 第一感覺是bug
• 第二感覺是偶發事件
• 筆者多次測試,并反復Review代碼,都沒發現其中的蹊蹺
• 參考筆者博文《手搓》TaskFactory帶你安全的起飛

var options = new ReduceOptions { ConcurrencyLevel = 4 };
var factory = new ConcurrentTaskFactory(options);
_output.WriteLine($"begin {DateTime.Now:HH:mm:ss.fff}");
Stopwatch sw = Stopwatch.StartNew();
List<Task<Product>> tasks = new(100);
for (int i = 0; i < 100; i++)
{
    var id = i;
    var task = factory.StartTask(() => GetProductAsync(id));
    tasks.Add(task);
}
var products = await Task.WhenAll(tasks);
sw.Stop();
_output.WriteLine($"end {DateTime.Now:HH:mm:ss.fff}, Elapsed {sw.ElapsedMilliseconds}");
Assert.NotNull(products);
Assert.Equal(100, products.Length);

// begin 10:20:45.317
// Thread36 GetProductAsync(0),10:20:45.487
// Thread11 GetProductAsync(2),10:20:45.487
// Thread8 GetProductAsync(3),10:20:45.487
// Thread35 GetProductAsync(1),10:20:45.487
// Thread11 GetProductAsync(6),10:20:45.614
// Thread35 GetProductAsync(4),10:20:45.614
// Thread8 GetProductAsync(7),10:20:45.614
// Thread36 GetProductAsync(5),10:20:45.614
// Thread35 GetProductAsync(9),10:20:45.742
// Thread8 GetProductAsync(8),10:20:45.742
// Thread41 GetProductAsync(11),10:20:45.742
// Thread37 GetProductAsync(10),10:20:45.742
// Thread37 GetProductAsync(12),10:20:45.869
// Thread8 GetProductAsync(14),10:20:45.869
// Thread11 GetProductAsync(13),10:20:45.869
// Thread41 GetProductAsync(15),10:20:45.869
// Thread8 GetProductAsync(18),10:20:45.997
// Thread35 GetProductAsync(17),10:20:45.997
// Thread41 GetProductAsync(19),10:20:45.997
// Thread11 GetProductAsync(16),10:20:45.997
// Thread11 GetProductAsync(20),10:20:46.125
// Thread8 GetProductAsync(22),10:20:46.125
// Thread35 GetProductAsync(21),10:20:46.125
// Thread41 GetProductAsync(23),10:20:46.125
// Thread37 GetProductAsync(27),10:20:46.253
// Thread41 GetProductAsync(26),10:20:46.253
// Thread35 GetProductAsync(25),10:20:46.253
// Thread11 GetProductAsync(24),10:20:46.253
// Thread35 GetProductAsync(29),10:20:46.381
// Thread41 GetProductAsync(28),10:20:46.381
// Thread11 GetProductAsync(31),10:20:46.381
// Thread37 GetProductAsync(30),10:20:46.381
// Thread8 GetProductAsync(32),10:20:46.507
// Thread37 GetProductAsync(34),10:20:46.507
// Thread41 GetProductAsync(35),10:20:46.507
// Thread11 GetProductAsync(33),10:20:46.507
// Thread37 GetProductAsync(39),10:20:46.635
// Thread11 GetProductAsync(37),10:20:46.635
// Thread41 GetProductAsync(36),10:20:46.635
// Thread35 GetProductAsync(38),10:20:46.635
// Thread41 GetProductAsync(40),10:20:46.763
// Thread37 GetProductAsync(41),10:20:46.763
// Thread35 GetProductAsync(42),10:20:46.763
// Thread11 GetProductAsync(43),10:20:46.763
// Thread41 GetProductAsync(47),10:20:46.891
// Thread8 GetProductAsync(44),10:20:46.891
// Thread11 GetProductAsync(46),10:20:46.891
// Thread37 GetProductAsync(45),10:20:46.891
// Thread37 GetProductAsync(51),10:20:47.018
// Thread8 GetProductAsync(49),10:20:47.018
// Thread41 GetProductAsync(48),10:20:47.018
// Thread11 GetProductAsync(50),10:20:47.018
// Thread41 GetProductAsync(55),10:20:47.146
// Thread11 GetProductAsync(54),10:20:47.146
// Thread8 GetProductAsync(52),10:20:47.146
// Thread37 GetProductAsync(53),10:20:47.146
// Thread11 GetProductAsync(59),10:20:47.274
// Thread8 GetProductAsync(58),10:20:47.274
// Thread37 GetProductAsync(56),10:20:47.274
// Thread41 GetProductAsync(57),10:20:47.274
// Thread41 GetProductAsync(62),10:20:47.402
// Thread11 GetProductAsync(63),10:20:47.402
// Thread37 GetProductAsync(60),10:20:47.402
// Thread8 GetProductAsync(61),10:20:47.402
// Thread41 GetProductAsync(66),10:20:47.530
// Thread88 GetProductAsync(64),10:20:47.530
// Thread35 GetProductAsync(65),10:20:47.530
// Thread11 GetProductAsync(67),10:20:47.530
// Thread11 GetProductAsync(68),10:20:47.658
// Thread41 GetProductAsync(70),10:20:47.658
// Thread8 GetProductAsync(69),10:20:47.658
// Thread35 GetProductAsync(71),10:20:47.658
// Thread41 GetProductAsync(74),10:20:47.786
// Thread95 GetProductAsync(75),10:20:47.786
// Thread35 GetProductAsync(73),10:20:47.786
// Thread88 GetProductAsync(72),10:20:47.786
// Thread95 GetProductAsync(78),10:20:47.914
// Thread41 GetProductAsync(77),10:20:47.914
// Thread88 GetProductAsync(79),10:20:47.914
// Thread8 GetProductAsync(76),10:20:47.914
// Thread95 GetProductAsync(80),10:20:48.042
// Thread41 GetProductAsync(83),10:20:48.042
// Thread8 GetProductAsync(82),10:20:48.042
// Thread35 GetProductAsync(81),10:20:48.042
// Thread95 GetProductAsync(84),10:20:48.170
// Thread35 GetProductAsync(86),10:20:48.170
// Thread41 GetProductAsync(85),10:20:48.170
// Thread8 GetProductAsync(87),10:20:48.170
// Thread11 GetProductAsync(90),10:20:48.297
// Thread88 GetProductAsync(88),10:20:48.297
// Thread8 GetProductAsync(89),10:20:48.297
// Thread35 GetProductAsync(91),10:20:48.297
// Thread11 GetProductAsync(95),10:20:48.425
// Thread8 GetProductAsync(94),10:20:48.425
// Thread41 GetProductAsync(93),10:20:48.425
// Thread35 GetProductAsync(92),10:20:48.425
// Thread41 GetProductAsync(98),10:20:48.553
// Thread35 GetProductAsync(99),10:20:48.553
// Thread8 GetProductAsync(97),10:20:48.553
// Thread88 GetProductAsync(96),10:20:48.553
// end 10:20:48.553, Elapsed 3235

三、進一步優化的空間

• 傍晚在小區邊的小湖散步就反復思考異步并發的問題
• 如果把異步線程等同同步線程,在異步線程未執行完就激活了新的線程
• 由于激活前這個過程足夠快導致線程疊加
• 造成開局即是高潮的"假象"

1. 《手搓》線程池核心代碼

• 異步時_processor.Run的耗時可以忽略不計
• 如何能實現同步和異步同樣的效果

while (true)
{
    if (_processor.Run())
    {
        _pool.Increment();
    }
    else
    {
        await Task.Delay(_reduceTime, CancellationToken.None)
            .ConfigureAwait(false);
    }
    if (token.IsCancellationRequested)
        break;
}

2. 以上想明白后就好優化了

• 優化后的代碼
• 把_processor.Run()拆分processor.TryTake(out var item)和processor.Run(ref item)
• 先發現任務,激活線程再執行
• 有人可能會說,你這代碼邏輯有問題啊
• 發現任務也只是證明本線程有事可干,激活新線程明顯浪費資源啊
• 這點確實應該說一下
• 現實經驗告訴我們,如果發現1只蟑螂,很可能你家已經成為了蟑螂窩
• 就算激活一個線程沒事可干消耗也不大,TryTake返回false就直接走線程回收邏輯了

while (true)
{
    if (processor.TryTake(out var item))
    {
        _pool.Increment();
        processor.Run(ref item);
    }
    else
    {
        await Task.Delay(_reduceTime, CancellationToken.None)
            .ConfigureAwait(false);
    }
    if (token.IsCancellationRequested)
        break;
}

3. 優化后再把第一個同步Case重跑一遍

• ConcurrencyLevel設置為10
• 同步方法也實現了一發入魂,開局即是高潮
• 必需承認完美的指數遞進關系很有數學美
• 但筆者做為一個碼奴,追求程序性能才是終極目標

var options = new ReduceOptions { ConcurrencyLevel = 10 };
var scheduler = new ConcurrentTaskScheduler(options);
var factory = new TaskFactory(scheduler);
var jobService = new ConcurrentJobService(scheduler, options);
jobService.Start();
Start(factory);

private void Start(TaskFactory factory)
{
    for (int i = 1; i < 10; i++)
    {
        for (int j = 1; j < 10; j++)
        {
            int a = i, b = j;
            factory.StartNew(() => Multiply(a, b));
        }
    }
}
public int Multiply(int a, int b)
{
    var result = a * b;
    _output.WriteLine($"{a} x {b} = {result},{DateTime.Now:HH:mm:ss.fff}");
    Thread.Sleep(100);
    return result;
}
// 1 x 6 = 6,09:15:35.332
// 1 x 3 = 3,09:15:35.332
// 1 x 2 = 2,09:15:35.332
// 1 x 4 = 4,09:15:35.332
// 1 x 5 = 5,09:15:35.332
// 1 x 9 = 9,09:15:35.333
// 1 x 7 = 7,09:15:35.333
// 1 x 8 = 8,09:15:35.333
// 1 x 1 = 1,09:15:35.332
// 2 x 1 = 2,09:15:35.333
// 2 x 6 = 12,09:15:35.443
// 2 x 3 = 6,09:15:35.443
// 2 x 7 = 14,09:15:35.443
// 2 x 4 = 8,09:15:35.443
// 2 x 8 = 16,09:15:35.443
// 2 x 5 = 10,09:15:35.443
// 2 x 2 = 4,09:15:35.443
// 2 x 9 = 18,09:15:35.443
// 3 x 1 = 3,09:15:35.443
// 3 x 2 = 6,09:15:35.444
// 4 x 3 = 12,09:15:35.554
// 3 x 4 = 12,09:15:35.554
// 3 x 5 = 15,09:15:35.554
// 3 x 3 = 9,09:15:35.554
// 3 x 7 = 21,09:15:35.554
// 3 x 9 = 27,09:15:35.554
// 4 x 1 = 4,09:15:35.554
// 3 x 8 = 24,09:15:35.554
// 3 x 6 = 18,09:15:35.554
// 4 x 2 = 8,09:15:35.554
// 4 x 6 = 24,09:15:35.666
// 5 x 4 = 20,09:15:35.666
// 4 x 8 = 32,09:15:35.666
// 4 x 9 = 36,09:15:35.666
// 4 x 7 = 28,09:15:35.666
// 5 x 3 = 15,09:15:35.666
// 4 x 4 = 16,09:15:35.666
// 5 x 2 = 10,09:15:35.666
// 5 x 1 = 5,09:15:35.666
// 4 x 5 = 20,09:15:35.666
// 5 x 6 = 30,09:15:35.777
// 6 x 1 = 6,09:15:35.777
// 6 x 5 = 30,09:15:35.777
// 6 x 4 = 24,09:15:35.777
// 5 x 9 = 45,09:15:35.777
// 5 x 5 = 25,09:15:35.777
// 6 x 2 = 12,09:15:35.777
// 5 x 8 = 40,09:15:35.777
// 5 x 7 = 35,09:15:35.777
// 6 x 3 = 18,09:15:35.777
// 6 x 9 = 54,09:15:35.888
// 6 x 8 = 48,09:15:35.888
// 7 x 1 = 7,09:15:35.888
// 6 x 7 = 42,09:15:35.888
// 6 x 6 = 36,09:15:35.888
// 7 x 6 = 42,09:15:35.888
// 7 x 4 = 28,09:15:35.888
// 7 x 3 = 21,09:15:35.888
// 7 x 5 = 35,09:15:35.888
// 7 x 2 = 14,09:15:35.888
// 7 x 8 = 56,09:15:36.000
// 7 x 7 = 49,09:15:36.000
// 7 x 9 = 63,09:15:36.000
// 8 x 1 = 8,09:15:36.000
// 8 x 7 = 56,09:15:36.000
// 8 x 2 = 16,09:15:36.000
// 8 x 3 = 24,09:15:36.000
// 8 x 5 = 40,09:15:36.000
// 8 x 6 = 48,09:15:36.000
// 8 x 4 = 32,09:15:36.000
// 8 x 8 = 64,09:15:36.112
// 8 x 9 = 72,09:15:36.112
// 9 x 2 = 18,09:15:36.112
// 9 x 1 = 9,09:15:36.112
// 9 x 3 = 27,09:15:36.112
// 9 x 4 = 36,09:15:36.112
// 9 x 5 = 45,09:15:36.112
// 9 x 7 = 63,09:15:36.112
// 9 x 6 = 54,09:15:36.112
// 9 x 8 = 72,09:15:36.112
// 9 x 9 = 81,09:15:36.223

4. 線程池執行同步方法開局即是高潮重要嗎

• 當然重要,這就是冷啟動的速度問題
• 如果用TaskFactory對同步方法做并發操作就更顯得重要了
• GetProduct同步方法執行1次耗時0.1秒
• ConcurrencyLevel設置為7
• 現在優化后獲取7條只要0.1秒
• 如果冷啟動按指數遞進關系至少要0.3秒多
• 這就快了3倍
• 如果ConcurrencyLevel越大,效果就越明顯

var options = new ReduceOptions { ConcurrencyLevel = 7 };
var factory = new ConcurrentTaskFactory(options);
_output.WriteLine($"begin {DateTime.Now:HH:mm:ss.fff}");
Stopwatch sw = Stopwatch.StartNew();
List<Task<Product>> tasks = new(7);
for (int i = 0; i < 7; i++)
{
    var id = i;
    var task = factory.StartNew(() => GetProduct(id));
    tasks.Add(task);
}
var products = await Task.WhenAll(tasks);
sw.Stop();
_output.WriteLine($"end {DateTime.Now:HH:mm:ss.fff}, Elapsed {sw.ElapsedMilliseconds}");

internal Product GetProduct(int id)
{
    Thread.Sleep(100);
    _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} GetProductAsync({id}),{DateTime.Now:HH:mm:ss.fff}");
    return new(id);
}

// begin 09:52:02.916
// Thread36 GetProductAsync(5),09:52:03.079
// Thread32 GetProductAsync(1),09:52:03.079
// Thread33 GetProductAsync(2),09:52:03.079
// Thread37 GetProductAsync(6),09:52:03.079
// Thread34 GetProductAsync(3),09:52:03.079
// Thread35 GetProductAsync(4),09:52:03.079
// Thread8 GetProductAsync(0),09:52:03.079
// end 09:52:03.079, Elapsed 162

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

如果大家喜歡請動(dong)動(dong)您發財(cai)的(de)小手手幫忙(mang)點(dian)一下Star,謝(xie)謝(xie)！！！