非 Win Form C# 应用程序中的同步事件处理程序线程执行答案

【问题标题】：Synchronous Event Handler thread execution in non-Win Form C# app非 Win Form C# 应用程序中的同步事件处理程序线程执行
【发布时间】：2012-05-04 07:37:02
【问题描述】：

我正在一个接一个地运行大量时间\CPU 密集型进程（TimeExpensive 类型）。主线程 (A) 在另一个线程 (B) 中异步启动 TimeExpensive 进程并变为非活动状态。在完成时，线程 B 同步触发调用者的完成处理程序并在线程 B 中启动下一个 TimeExpensive 进程。创建了一个新线程 (C)，但在启动 C 后，B 完成。因此，对于 n 个进程，创建了 n 个线程，并且大多数时候，它们并不共存。

人们可能希望它以线性单线程方式实现，但 TimeExpensive 是由第三方实现的，并且在运行时会使用所有系统内核并运行数小时。

//This will run as console app 
class Program
{
    static void Main(string[] args)
    {
        new Program().StartJobs();
    }

    void StartJobs()
    {
        Main mainJob = new Main();
        mainJob.MainCompletionEvent += 
                     new Action<object, EventArgs>(mainJob_MainCompletionEvent);
        mainJob.Start();
    }

    void mainJob_MainCompletionEvent(object sender, EventArgs e)
    {
        //if(success) Environment.Exit(0);
    }

}

class Main
{
    int processCounter = 0;
    public event Action<object, EventArgs> MainCompletionEvent;
    public void Start()
    {
        //...do other important tasks here...
        processCounter++;
        TimeExpensive te = new TimeExpensive();
        te.CompletionEvent += new Action(TimeExpensive_CompletionHandler);
        Thread aThread = new Thread(te.Run);
        aThread.IsBackground = false;
        aThread.Name = "TimeExpensive Thread: " + processCounter;
        aThread.Start();
    }

    void TimeExpensive_CompletionHandler()
    {
        Console.WriteLine("current Thread Name: " + Thread.CurrentThread.Name);
        //Start another Process In Background if
        if (processCounter < 5)
        {
            Start();
        }
        else
        {
            Console.ReadKey();
            if (JobCompletionEvent != null)
                JobCompletionEvent(this, new EventArgs());
        }
    }
}

class TimeExpensive
{
    public event Action CompletionEvent;

    public void Run()
    {
        //doing time expensive task
        //...

        //when finish Notify completion Handler...
        if (CompletionEvent != null)
        {
            CompletionEvent();
        }
    }
}

//Output
current Thread Name: TimeExpensive Thread: 1
current Thread Name: TimeExpensive Thread: 2
current Thread Name: TimeExpensive Thread: 3
current Thread Name: TimeExpensive Thread: 4
current Thread Name: TimeExpensive Thread: 5

上面的实现模仿了我描述的行为。困扰我的事情是事件处理程序同步运行，直到下一个线程启动，在此期间，它正在执行许多它不是为之设计的任务。

不确定这是否好，有没有办法可以在线程 B 的完成处理程序中返回线程 A？还是我应该更好地使用另一个 delegate.BeginInvoke 启动事件处理程序执行？

我希望用简单而安全的方法来做这件事。任何帮助是极大的赞赏。

P.S 我读了很多帖子，但没有人能很好地处理这种情况。

编辑

添加了静态 main 以显示如何在控制台应用程序中启动此代码。请记住，还可以创建 UI 来启动“主要”工作。它肯定会创建 BackgroundWorker 线程来创建 mainJob 对象并运行它。谢谢！

【问题讨论】：

'它使用所有系统核心'它可能以一种不寻常的方式实现，但它似乎做了很多有用的工作？我不认为在 OnCommpleteion 处理程序中工作是必然邪恶的——它是工作，只是另一个函数调用。无论如何，我不明白对 A、B、C 使用不同的线程。为什么这不能按顺序运行？无论如何，如果需要进行任何更改，我会转储线程并为 A、B、C 提交池任务。
使用多核是加分项，尤其是当您拥有专用服务器时。 TimeExpensive 过程由第三方执行以运行 Async。我只使用了一个 dll。
在另一个线程中运行 TimeExpensive 的另一个优点是在 TimeExpensive 上间接调用 CancelAsync。中止正在运行的 TE 线程，尤其是在它执行 IO 时是不安全的。
这是一个控制台应用程序吗？如果是这样，你能展示你的 Main 方法吗？
@NicholasButler 我刚刚添加了控制台应用程序代码。查看我的编辑 cmets。

标签： c# multithreading asynchronous event-handling

【解决方案1】：

有没有办法在线程 B 的完成处理程序中返回线程 A？

不，您没有合适的管道将调用从一个线程编组到另一个线程。这种管道由 GUI 应用程序的主线程提供。出于必要，用户界面基本上是线程不安全的。支持这种封送处理的 UI 线程有几个实现细节。它在producer/consumer 线程模型的典型实现中充当消费者。

这需要一个线程安全的队列和一个读取队列的消费者循环。您可能会将其识别为 Windows GUI 应用程序中的消息队列。使用调用 GetMessage 的消息循环来读取通知并对其采取行动。编组调用现在很简单，您只需将消息发布到消息队列，UI 线程读取它并执行请求。 Posting 由 Winforms 的 Control.BeginInvoke 和 WPF 的 Dispatcher.BeginInvoke 实现。

您当然可以自己实现这种同步机制。 .NET 4 BlockingCollection 类使它变得容易。但是做请记住，您必须从根本上改变线程 A 的执行方式。对发布到队列的请求保持响应很重要。像 BackgroundWorker 这样的类试图解决的问题。请记住，GUI 消息循环的存在是因为它是必要的，UI 不是线程安全的。控制台应用程序（通常）没有同样的负担，控制台是线程安全的。

【讨论】：

+1 将我指向 BlockingCollection 并回答了主要问题。

【解决方案2】：

您遇到的问题是由于正确穿线非常困难。我用演员编写了你的例子：

type Actor<'a> = MailboxProcessor<'a>

type SupMsg = WaitForDone of AsyncReplyChannel<string>
type ProgramState = RunNumber of int * Actor<WorkerMsg> option
and WorkerMsg = Begin of Id * AsyncReplyChannel<string>
and Id = int

let startComputation () = Actor.Start(fun inbox ->
  async { 
    let! Begin(id, chan) = inbox.Receive()
    printfn "Running Computation"
    do! Async.Sleep(20) // zZz
    chan.Reply(sprintf "'%i is done!'" id) })

let sup () = Actor.Start(fun inbox ->
  let rec loop state =
    async {
      match state with
      | RunNumber(_, None) -> return! loop <| RunNumber(1, Some(startComputation ()))
      | RunNumber(run, Some(active)) ->
        let! completed = active.PostAndAsyncReply(fun chan -> Begin(run, chan))
        printfn "sup observed: %s" completed
        let active' = Some(startComputation ())
        if run <> 5 then return! loop <| RunNumber(run + 1, active')
        else return! isDone () }
  and isDone () =
    async {
      let! WaitForDone(chan) = inbox.Receive()
      return chan.Reply("all done") }

  loop <| RunNumber(0, None))

[<EntryPoint>]
let main args =
  printfn "%s" <| (sup ()).PostAndReply(fun chan -> WaitForDone(chan))
  0

作为输出：

> main();;
Running Computation
sup observed: '1 is done!'
Running Computation
sup observed: '2 is done!'
Running Computation
sup observed: '3 is done!'
Running Computation
sup observed: '4 is done!'
Running Computation
sup observed: '5 is done!'
all done
val it : int = 0

如您所见，跨线程通信变得轻而易举。如果您的库是第三方库，那么很容易将Async.Sleep(20) 替换为对该库的调用。

【讨论】：