我想要一个特定的线程,任务队列并在该单独的线程中处理任务。应用程序将根据用户的使用情况创建任务并将其排队到任务队列中。然后单独的线程处理任务。即使队列为空,保持线程活动并使用它来处理排队任务也至关重要。
我尝试过几种实现TaskScheduler
with BlockingCollection
并将并发限制为仅一个线程,但当队列变空并且任务由其他线程处理时,线程似乎被释放。
您能否至少向我介绍一些如何实现这一目标的资源?
太长了;博士
尝试限制一个特定线程来处理动态添加到队列中的任务。
Edit1:
这是使用 WCF 和 .NET Framework 4.6 的实验性 Web 应用程序。在 WCF 库中,我试图通过一个线程处理任务来实现这一行为。这一线程必须使用外部 dll 库初始化 prolog,然后使用 prolog。如果进程中使用其他线程,库会抛出异常AccessViolationException
。我做了一些研究,这很可能是因为该库中的线程管理不善。我的实现是到处都有锁并且它有效。我现在正在尝试重新实现并使其异步,这样我就不会通过锁定来阻止主线程。
我不在电脑前,但今天晚些时候回家时我提供了一些代码。
你的方法看起来不错,所以你可能只是犯了一些微小的愚蠢错误。
实际上很容易进行简单的定制TaskScheduler
。对于你的情况:
void Main()
{
var cts = new CancellationTokenSource();
var myTs = new SingleThreadTaskScheduler(cts.Token);
myTs.Schedule(() =>
{ Print("Init start"); Thread.Sleep(1000); Print("Init done"); });
myTs.Schedule(() => Print("Work 1"));
myTs.Schedule(() => Print("Work 2"));
myTs.Schedule(() => Print("Work 3"));
var lastOne = myTs.Schedule(() => Print("Work 4"));
Print("Starting TS");
myTs.Start();
// Wait for all of them to complete...
lastOne.GetAwaiter().GetResult();
Thread.Sleep(1000);
// And try to schedule another
myTs.Schedule(() => Print("After emptied")).GetAwaiter().GetResult();
// And shutdown; it's also pretty useful to have the
// TaskScheduler return a "complete task" to await
myTs.Complete();
Print("On main thread again");
}
void Print(string str)
{
Console.WriteLine("{0}: {1}", Thread.CurrentThread.ManagedThreadId, str);
Thread.Sleep(100);
}
public sealed class SingleThreadTaskScheduler : TaskScheduler
{
[ThreadStatic]
private static bool isExecuting;
private readonly CancellationToken cancellationToken;
private readonly BlockingCollection<Task> taskQueue;
public SingleThreadTaskScheduler(CancellationToken cancellationToken)
{
this.cancellationToken = cancellationToken;
this.taskQueue = new BlockingCollection<Task>();
}
public void Start()
{
new Thread(RunOnCurrentThread) { Name = "STTS Thread" }.Start();
}
// Just a helper for the sample code
public Task Schedule(Action action)
{
return
Task.Factory.StartNew
(
action,
CancellationToken.None,
TaskCreationOptions.None,
this
);
}
// You can have this public if you want - just make sure to hide it
private void RunOnCurrentThread()
{
isExecuting = true;
try
{
foreach (var task in taskQueue.GetConsumingEnumerable(cancellationToken))
{
TryExecuteTask(task);
}
}
catch (OperationCanceledException)
{ }
finally
{
isExecuting = false;
}
}
// Signaling this allows the task scheduler to finish after all tasks complete
public void Complete() { taskQueue.CompleteAdding(); }
protected override IEnumerable<Task> GetScheduledTasks() { return null; }
protected override void QueueTask(Task task)
{
try
{
taskQueue.Add(task, cancellationToken);
}
catch (OperationCanceledException)
{ }
}
protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
{
// We'd need to remove the task from queue if it was already queued.
// That would be too hard.
if (taskWasPreviouslyQueued) return false;
return isExecuting && TryExecuteTask(task);
}
}
修改它非常容易,让您可以完全控制任务调度程序实际执行任务的位置 - 事实上,我已经从我之前使用过的任务调度程序中改编了它,它只具有RunOnCurrentThread
方法公开。
对于您的情况,您总是想坚持只使用一个线程,方法是SingleThreadTaskScheduler
可能更好。虽然这也有它的优点:
// On a new thread
try
{
InitializeProlog();
try
{
myTs.RunOnCurrentThread();
}
finally
{
ReleaseProlog();
}
}
catch (Exception ex)
{
// The global handler
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)