c# - 是否总是可以使用 Task 强制一个新线程？

Question

每次Task.Factory.StartNew调用我都试图创建一个新线程。问题是如何在不抛出异常的情况下运行下面的代码：

static void Main(string[] args)
{
    int firstThreadId = 0;

    Task.Factory.StartNew(() => firstThreadId = Thread.CurrentThread.ManagedThreadId);

    for (int i = 0; i < 100; i++)
    {
        Task.Factory.StartNew(() =>
        {
            while (true)
            {
                Thread.Sleep(1000);
                if (firstThreadId == Thread.CurrentThread.ManagedThreadId)
                    throw new Exception("The first thread is reused.");
            }
        });
    }
    Console.Read();
}

编辑：如果您评论第一条语句，新代码for没有问题。但是如果你有它，哇，消息“线程重用”被写入控制台。你能解释一下吗，因为我真的很困惑。

static void Main(string[] args)
{
    ConcurrentDictionary<int, int> startedThreads = new ConcurrentDictionary<int, int>();

    for (int i = 0; i < 10; i++)
    {
        Task.Factory.StartNew(() =>
        {
            Task.Factory.StartNew(() =>
            {
                startedThreads.AddOrUpdate(Thread.CurrentThread.ManagedThreadId,
                    Thread.CurrentThread.ManagedThreadId, (a, b) => b);
            }, TaskCreationOptions.LongRunning);

            for (int j = 0; j < 100; j++)
            {
                Task.Factory.StartNew(() =>
                {
                    while (true)
                    {
                        Thread.Sleep(10);
                        if (startedThreads.ContainsKey(
                            Thread.CurrentThread.ManagedThreadId))
                                Console.WriteLine("Thread reused");
                    }
                }, TaskCreationOptions.LongRunning);
            }
        });
    }

    Console.Read();
}

score 23 · Accepted Answer

如果您TaskCreationOptions.LongRunning在启动任务时指定，则会向调度程序提供提示，默认调度程序将其作为为任务创建新线程的指示符。

这只是一个提示 - 我不确定我是否会依赖它......但我没有看到任何使用默认调度程序的反例。

score 6 · Accepted Answer

添加到 Jon Skeet 的答案中，如果您想保证每次都创建一个新线程，您可以编写自己的TaskScheduler创建一个新线程。

score 3 · Accepted Answer

您好，谢谢大家的回答。你们都得到了+1。所有建议的解决方案都不适用于我的情况。问题是当你休眠一个线程时，它会在某个时间点被重用。上面的人建议：

使用 LongRunning => 如果您有嵌套/子任务，这将不起作用
自定义任务调度程序 => 我尝试编写自己的任务调度程序，也尝试了这个 ThreadPerTaskScheduler，它也不起作用。
使用纯线程 => 仍然失败...
你也可以在Multithreading.Scheduler github 上查看这个项目

我的解决方案

我不喜欢它，但它有效。基本上我阻塞了线程，所以它不能被重用。下面是扩展方法和一个工作示例。再次谢谢你。

https://gist.github.com/4150635

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication
{
    public static class ThreadExtensions
    {
        /// <summary>
        /// Blocks the current thread for a period of time so that the thread cannot be reused by the threadpool.
        /// </summary>
        public static void Block(this Thread thread, int millisecondsTimeout)
        {
            new WakeSleepClass(millisecondsTimeout).SleepThread();
        }

        /// <summary>
        /// Blocks the current thread so that the thread cannot be reused by the threadpool.
        /// </summary>
        public static void Block(this Thread thread)
        {
            new WakeSleepClass().SleepThread();
        }

        /// <summary>
        /// Blocks the current thread for a period of time so that the thread cannot be reused by the threadpool.
        /// </summary>
        public static void Block(this Thread thread, TimeSpan timeout)
        {
            new WakeSleepClass(timeout).SleepThread();
        }

        class WakeSleepClass
        {
            bool locked = true;
            readonly TimerDisposer timerDisposer = new TimerDisposer();

            public WakeSleepClass(int sleepTime)
            {
                var timer = new Timer(WakeThread, timerDisposer, sleepTime, sleepTime);
                timerDisposer.InternalTimer = timer;
            }

            public WakeSleepClass(TimeSpan sleepTime)
            {
                var timer = new Timer(WakeThread, timerDisposer, sleepTime, sleepTime);
                timerDisposer.InternalTimer = timer;
            }

            public WakeSleepClass()
            {
                var timer = new Timer(WakeThread, timerDisposer, Timeout.Infinite, Timeout.Infinite);
                timerDisposer.InternalTimer = timer;
            }

            public void SleepThread()
            {
                while (locked)
                    lock (timerDisposer) Monitor.Wait(timerDisposer);
                locked = true;
            }

            public void WakeThread(object key)
            {
                locked = false;
                lock (key) Monitor.Pulse(key);
                ((TimerDisposer)key).InternalTimer.Dispose();
            }

            class TimerDisposer
            {
                public Timer InternalTimer { get; set; }
            }
        }
    }

    class Program
    {
        private static readonly Queue<CancellationTokenSource> tokenSourceQueue = new Queue<CancellationTokenSource>();
        static void Main(string[] args)
        {
            CancellationTokenSource tokenSource = new CancellationTokenSource();
            tokenSourceQueue.Enqueue(tokenSource);

            ConcurrentDictionary<int, int> startedThreads = new ConcurrentDictionary<int, int>();
            for (int i = 0; i < 10; i++)
            {
                Thread.Sleep(1000);
                Task.Factory.StartNew(() =>
                {
                    startedThreads.AddOrUpdate(Thread.CurrentThread.ManagedThreadId, Thread.CurrentThread.ManagedThreadId, (a, b) => b);
                    for (int j = 0; j < 50; j++)
                        Task.Factory.StartNew(() => startedThreads.AddOrUpdate(Thread.CurrentThread.ManagedThreadId, Thread.CurrentThread.ManagedThreadId, (a, b) => b));

                    for (int j = 0; j < 50; j++)
                    {
                        Task.Factory.StartNew(() =>
                        {
                            while (!tokenSource.Token.IsCancellationRequested)
                            {
                                if (startedThreads.ContainsKey(Thread.CurrentThread.ManagedThreadId)) Console.WriteLine("Thread reused");
                                Thread.CurrentThread.Block(10);
                                if (startedThreads.ContainsKey(Thread.CurrentThread.ManagedThreadId)) Console.WriteLine("Thread reused");
                            }
                        }, tokenSource.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default)
                        .ContinueWith(task =>
                        {
                            WriteExceptions(task.Exception);
                            Console.WriteLine("-----------------------------");
                        }, TaskContinuationOptions.OnlyOnFaulted);
                    }
                    Thread.CurrentThread.Block();
                }, tokenSource.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default)
                .ContinueWith(task =>
                {
                    WriteExceptions(task.Exception);
                    Console.WriteLine("-----------------------------");
                }, TaskContinuationOptions.OnlyOnFaulted);
            }

            Console.Read();
        }

        private static void WriteExceptions(Exception ex)
        {
            Console.WriteLine(ex.Message);
            if (ex.InnerException != null)
                WriteExceptions(ex.InnerException);
        }
    }
}

score 3 · Accepted Answer

试试这个：

var taskCompletionSource = new TaskCompletionSource<bool>();
Thread t = new Thread(() =>
{
    try
    {
        Operation();
        taskCompletionSource.TrySetResult(true);
    }
    catch (Exception e)
    {
        taskCompletionSource.TrySetException(e);
    }
});
void Operation()
{
    // Some work in thread
}
t.Start();
await taskCompletionSource.Task;

您还可以为 Action、Func 等编写扩展方法。例如：

public static Task RunInThread(
    this Action action,
    Action<Thread> initThreadAction = null)
{
    TaskCompletionSource<bool> taskCompletionSource = new TaskCompletionSource<bool>();

    Thread thread = new Thread(() =>
    {
        try
        {
            action();
            taskCompletionSource.TrySetResult(true);
        }
        catch (Exception e)
        {
            taskCompletionSource.TrySetException(e);
        }
    });
    initThreadAction?.Invoke(thread);
    thread.Start();

    return taskCompletionSource.Task;
}

或者

public static Task<TResult> RunInThread<T1, T2, TResult>(
    this Func<T1, T2, TResult> function,
    T1 param1,
    T2 param2,
    Action<Thread> initThreadAction = null)
{
    TaskCompletionSource<TResult> taskCompletionSource = new TaskCompletionSource<TResult>();

    Thread thread = new Thread(() =>
    {
        try
        {
            TResult result = function(param1, param2);
            taskCompletionSource.TrySetResult(result);
        }
        catch (Exception e)
        {
            taskCompletionSource.TrySetException(e);
        }
    });
    initThreadAction?.Invoke(thread);
    thread.Start();

    return taskCompletionSource.Task;
}

并像这样使用它：

var result = await some_function.RunInThread(param1, param2).ConfigureAwait(true);

score 2 · Accepted Answer

只需使用 new Thread() 启动线程，然后 Start() 他们

static void Main(string[] args)
{
    ConcurrentDictionary<int, int> startedThreads = new ConcurrentDictionary<int, int>();

    for (int i = 0; i < 10; i++)
    {
        new Thread(() =>
        {
            new Thread(() =>
            {
                startedThreads.AddOrUpdate(Thread.CurrentThread.ManagedThreadId, Thread.CurrentThread.ManagedThreadId, (a, b) => b);
            }).Start();

            for (int j = 0; j < 100; j++)
            {
                new Thread(() =>
                {
                    while (true)
                    {
                        Thread.Sleep(10);
                        if (startedThreads.ContainsKey(Thread.CurrentThread.ManagedThreadId))
                            Console.WriteLine("Thread reused");
                    }
                }).Start();
            }
        }).Start();
    }

    Console.Read();

}

任务应该由调度程序管理。Tasks 的整个想法是运行时将决定何时需要新线程。另一方面，如果您确实需要不同的线程，则代码中的其他内容可能是错误的，例如过度依赖 Thread.Sleep() 或线程本地存储。

正如所指出的，您可以创建自己的 TaskScheduler 并使用任务来创建线程，但是为什么要使用 Tasks 开始呢？

score 0 · Accepted Answer

这是一个TaskScheduler在每个任务的专用线程上执行任务的自定义：

public class ThreadPerTask_TaskScheduler : TaskScheduler
{
    protected override void QueueTask(Task task)
    {
        var thread = new Thread(() => TryExecuteTask(task));
        thread.IsBackground = true;
        thread.Start();
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return TryExecuteTask(task);
    }

    protected override IEnumerable<Task> GetScheduledTasks() { yield break; }
}

使用示例：

var parallelOptions = new ParallelOptions()
{
    MaxDegreeOfParallelism = 3,
    TaskScheduler = new ThreadPerTask_TaskScheduler()
};

Parallel.ForEach(Enumerable.Range(1, 10), parallelOptions, item =>
{
    Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff}" +
        $" [{Thread.CurrentThread.ManagedThreadId}]" +
        $" Processing #{item}" +
        (Thread.CurrentThread.IsBackground ? ", Background" : "") +
        (Thread.CurrentThread.IsThreadPoolThread ? ", ThreadPool" : ""));
    Thread.Sleep(1000); // Simulate CPU-bound work
});

输出：

20:38:56.770 [4] Processing #3, Background
20:38:56.770 [5] Processing #2, Background
20:38:56.770 [1] Processing #1
20:38:57.782 [1] Processing #4
20:38:57.783 [8] Processing #5, Background
20:38:57.783 [7] Processing #6, Background
20:38:58.783 [1] Processing #7
20:38:58.783 [10] Processing #8, Background
20:38:58.787 [9] Processing #9, Background
20:38:59.783 [1] Processing #10

在 Fiddle 上试试。

此自定义TaskScheduler也允许当前线程参与计算。这在上面的示例中通过[1]处理项目#1、#4和#7的线程进行了演示#10。如果您不希望发生这种情况，只需将其中的代码替换TryExecuteTaskInline为return false;。

另一个例子，以Task.Factory.StartNew方法为特色。在 100 个不同的线程上启动 100 个任务：

var oneThreadPerTask = new ThreadPerTask_TaskScheduler();
Task[] tasks = Enumerable.Range(1, 100).Select(_ =>
{
    return Task.Factory.StartNew(() =>
    {
        Thread.Sleep(1000); // Simulate long-running work
    }, default, TaskCreationOptions.None, oneThreadPerTask);
}).ToArray();

在这种情况下，当前线程没有参与工作，因为所有任务都是通过调用它们的方法在幕后启动的Start，而不是 RunSynchronously.

c# - 是否总是可以使用 Task 强制一个新线程？

6 回答 6

我的解决方案

Related

Reference