如何为同时运行的多个 ActionBlock 设置 CPU 优先级？答案

【问题标题】：How to set CPU priority for multiple ActionBlocks running at the same time?如何为同时运行的多个 ActionBlock 设置 CPU 优先级？
【发布时间】：2020-03-12 12:24:32
【问题描述】：

我有一堆ActionBlocks，每个人都在做不同的事情。

大佬处理数据，通过TransformBlock连续喂数据。
其他 3 个 ActionBlocks 只需在 3 个文本文件（日志）中写入行。

它有点工作，除了 3 个日志记录 ActionBlocks 仅在处理 ActionBlock 完成时才开始消耗数据（因此他们在程序结束时一次性写入所有日志记录信息）。

我想知道我是否可以影响这种行为，从而为日志记录ActionBlocks 提供更高的优先级？

感谢您的帮助。

代码示例：

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace dataflowtest
{
    class Program
    {
        const string chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
        static readonly IReadOnlyCollection<string> charsSets = Enumerable.Repeat(chars, 8).ToList().AsReadOnly();
        static readonly Random random = new Random();

        static event EventHandler<string> MessageGot;

        static async Task Main(string[] args)
        {
            var source = new TransformBlock<string, string>(GetMessage, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = -1, EnsureOrdered = false });
            var target = new ActionBlock<string>(Console.WriteLine);


            var programDir = Path.GetDirectoryName(System.Reflection.Assembly.GetEntryAssembly().GetName().CodeBase.Replace("file:///", ""));
            using var file1 = new StreamWriter(Path.Combine(programDir, "file1.txt"));
            using var file2 = new StreamWriter(Path.Combine(programDir, "file2.txt"));
            using var file3 = new StreamWriter(Path.Combine(programDir, "file3.txt"));

            var fileAction1 = new ActionBlock<string>(file1.WriteLineAsync);
            var fileAction2 = new ActionBlock<string>(file2.WriteLineAsync);
            var fileAction3 = new ActionBlock<string>(file3.WriteLineAsync);

            MessageGot += async (_, e) => await fileAction1.SendAsync(e);
            MessageGot += async (_, e) => await fileAction2.SendAsync(e);
            MessageGot += async (_, e) => await fileAction3.SendAsync(e);

            using (source.LinkTo(target, new DataflowLinkOptions { PropagateCompletion = true }))
            {
                for (int i = 0; i < 100; i++)
                {
                    await source.SendAsync(i.ToString() + '\t' + new string(charsSets.Select(s => s[random.Next(s.Length)]).ToArray()));
                }

                source.Complete();
                await target.Completion;
            }
        }

        private static async Task<string> GetMessage(string input)
        {
            int delay = random.Next(25, 6000);
            await Task.Delay(delay);
            string message = input.ToLowerInvariant() + '\t' + delay.ToString();

            MessageGot?.Invoke(null, message);

            return message;
        }
    }
}

【问题讨论】：

我怀疑如果你在非调试模式下运行会发现不同的结果。
不，文件在程序结束之前一直是空的，然后在 Release 或 Debug 中立即填满。
“我假设” - 你为什么这么假设？我们需要查看您的代码以提供帮助
“我假设”，因为这对我来说很有意义。我正在添加一个小代码示例，为我重现该问题。
MaxDegreeOfParallelism = -1是什么原因？

标签： c# tpl-dataflow thread-priority

【解决方案1】：

默认情况下，StreamWriter 将每 4,096 字节刷新一次缓冲区。您可能希望它在写入的每一行都刷新。所以不要这样：

var fileAction1 = new ActionBlock<string>(file1.WriteLineAsync);

...这样做：

var fileAction1 = new ActionBlock<string>(item =>
{
    file1.WriteLine(item);
    file1.Flush();
});

使用WriteLineAsync 代替WriteLine 没有任何好处，因为基础FileStream 尚未使用FileOptions.Asynchronous 选项打开。

【讨论】：