碼奴蔣祥機(jī)

《手搓》線程池實(shí)現(xiàn)了完美的指數(shù)遞進(jìn)關(guān)系 異步并發(fā)測(cè)試一發(fā)入魂,開局即是高潮帶來了困惑 沉著思考后連夜優(yōu)化

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

• ConcurrencyLevel設(shè)置為10
• 并發(fā)實(shí)現(xiàn)了完美的指數(shù)遞進(jìn)關(guān)系
• 當(dāng)時(shí)內(nèi)心還是得到了很大的滿足
• 第一批01:58:55.832是1個(gè)并發(fā)
• 第二批01:58:55.944是2個(gè)并發(fā)
• 第三批01:58:56.054是4個(gè)并發(fā)
• 第四批01:58:56.165是8個(gè)并發(fā)
• 01:58:56.276及以后一段時(shí)間達(dá)到并發(fā)上限10個(gè)
• 參考筆者博文《手搓》線程池

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

二、短暫的快樂被異步并發(fā)測(cè)試給打破了

• ConcurrencyLevel設(shè)置為4
• 一開始就是4個(gè)清晰可見的并發(fā)
• 一發(fā)入魂,開局即是高潮
• 第一感覺是bug
• 第二感覺是偶發(fā)事件
• 筆者多次測(cè)試,并反復(fù)Review代碼,都沒發(fā)現(xiàn)其中的蹊蹺
• 參考筆者博文《手搓》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

三、進(jìn)一步優(yōu)化的空間

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

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

• 異步時(shí)_processor.Run的耗時(shí)可以忽略不計(jì)
• 如何能實(shí)現(xiàn)同步和異步同樣的效果

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

2. 以上想明白后就好優(yōu)化了

• 優(yōu)化后的代碼
• 把_processor.Run()拆分processor.TryTake(out var item)和processor.Run(ref item)
• 先發(fā)現(xiàn)任務(wù),激活線程再執(zhí)行
• 有人可能會(huì)說,你這代碼邏輯有問題啊
• 發(fā)現(xiàn)任務(wù)也只是證明本線程有事可干,激活新線程明顯浪費(fèi)資源啊
• 這點(diǎn)確實(shí)應(yīng)該說一下
• 現(xiàn)實(shí)經(jīng)驗(yàn)告訴我們,如果發(fā)現(xiàn)1只蟑螂,很可能你家已經(jīng)成為了蟑螂窩
• 就算激活一個(gè)線程沒事可干消耗也不大,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. 優(yōu)化后再把第一個(gè)同步Case重跑一遍

• ConcurrencyLevel設(shè)置為10
• 同步方法也實(shí)現(xiàn)了一發(fā)入魂,開局即是高潮
• 必需承認(rèn)完美的指數(shù)遞進(jìn)關(guān)系很有數(shù)學(xué)美
• 但筆者做為一個(gè)碼奴,追求程序性能才是終極目標(biāo)

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. 線程池執(zhí)行同步方法開局即是高潮重要嗎

• 當(dāng)然重要,這就是冷啟動(dòng)的速度問題
• 如果用TaskFactory對(duì)同步方法做并發(fā)操作就更顯得重要了
• GetProduct同步方法執(zhí)行1次耗時(shí)0.1秒
• ConcurrencyLevel設(shè)置為7
• 現(xiàn)在優(yōu)化后獲取7條只要0.1秒
• 如果冷啟動(dòng)按指數(shù)遞進(jìn)關(guān)系至少要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

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

如果大家喜歡請(qǐng)動(dòng)動(dòng)您發(fā)財(cái)?shù)男∈质謳兔c(diǎn)一下Star,謝謝！！！