您将如何在 .NET 中实现这个“WorkerChain”功能？

Question

编辑：我有点太晚了（？）我在第一次更新这个问题时发布的所有代码对于大多数读者来说太多 .实际上，我已经继续前进了，written a blog post about this topic 供任何愿意阅读它的人使用。

与此同时，我将原始问题留在原处，简要介绍一下我想解决的问题。

我还要注意，到目前为止，我发布的代码（在我的博客上）已经很好地经受住了测试。但我仍然对人们愿意就它的干净/稳健/性能*提供的任何和所有反馈感兴趣。

*_{我喜欢doesn't really mean what we think这个词，但我们开发人员一直都在使用它。}

---

原始问题

抱歉，问题标题含糊不清——不知道如何简洁地概括我在下面提出的问题。 （如果有编辑权限的人能想到一个更具描述性的标题，请随意更改。）

我需要的行为是这样的。我正在设想一个在其构造函数中接受单个委托任务的工作类（为简单起见，我将使其不可变——实例化后不能再添加任务）。我将此任务称为T。该类应该有一个简单的方法，比如GetToWork，它会表现出这种行为：

1. 如果工作人员当前没有运行T，那么它将立即开始运行。
2. 如果工作人员正在当前运行T，那么一旦完成，它将立即再次启动T。
3. GetToWork 可以在 worker 运行时调用任意次数 T；简单的规则是，在T 的任何执行期间，如果GetToWork 被调用至少一次，T 将在完成后再次运行（然后如果GetToWork 被调用而@ 987654334@ 正在运行那个时间，它会再次重复，等等）。

现在，使用布尔开关非常简单。但是这个类需要线程安全，我的意思是，上面的步骤 1 和 2 需要包含原子操作（至少我认为是这样）。

增加了一层复杂性。我需要一个“工人链”课程，它将由许多连接在一起的工人组成。一旦第一个工作人员完成，它实际上会在 after 工作人员上调用GetToWork；同时，如果它自己的GetToWork 已被调用，它也会重新启动自己。从逻辑上讲，在 链 上调用 GetToWork 与在链中 的第一个工人上调用 GetToWork 基本相同（我完全打算不公开该链的工人可访问）。

想象这个假设的“工人链”会如何表现的一种方法是将其与接力赛中的团队进行比较。假设有四个跑步者，W1 到W4，让链被称为C。如果我打电话给C.StartWork()，应该是这样的：

1. 如果W1在他的起点（即什么都不做），他将开始向W2跑。
2. 如果W1已经跑向W2（即执行他的任务），那么一旦他到达W2，他就会向W2发信号开始，立即返回到他的起点，因为StartWork已经被调用，再次开始向W2跑。
3. 当W1到达W2的起点时，他会立即回到自己的起点。
   1. 如果W2 只是坐在那里，他会立即开始跑向W3。
   2. 如果W2 已经跑向W3，那么W2 将在到达W3 并返回起点后再次运行。

上面的内容可能有点令人费解并且写得不好。但希望你能得到基本的想法。显然，这些工作线程将在自己的线程上运行。

另外，我猜这个功能可能已经存在于某个地方？如果是这样，一定告诉我！

Answer 1

使用信号量。每个worker都是一个线程，代码如下（伪代码）：

WHILE(TRUE)
    WAIT_FOR_SEMAPHORE(WORKER_ID) //The semaphore for the current worker
    RESET_SEMAPHORE(WORKER_ID)
    /* DO WORK */
    POST_SEMAPHORE(NEXT_WORKER_ID) //The semaphore for the next worker
END

非零信号量意味着有人向当前线程发出信号来完成工作。在其入口行中获得非零信号量后，它会重置信号量（标记为没有人发出信号），执行工作（同时可以再次发布信号量）并为下一个工作人员发布信号量。故事在下一个工人中重演。

Answer 2

一个幼稚的实现，您可能会从中获得一些好处。

注意：

据我了解，标量类型，r.e.控制执行的布尔标志，具有原子分配，使它们成为您在这种情况下需要/想要的线程安全。

涉及信号量和其他策略的可能性要复杂得多，但如果简单的话......

using System;
using System.Threading;

namespace FlaggedWorkerChain
{
    internal class Program
    {
        private static void Main(string[] args)
        {
            FlaggedChainedWorker innerWorker = new FlaggedChainedWorker("inner", () => Thread.Sleep(1000), null);
            FlaggedChainedWorker outerWorker = new FlaggedChainedWorker("outer", () => Thread.Sleep(500), innerWorker);

            Thread t = new Thread(outerWorker.GetToWork);
            t.Start();

            // flag outer to do work again
            outerWorker.GetToWork();

            Console.WriteLine("press the any key");
            Console.ReadKey();
        }
    }

    public sealed class FlaggedChainedWorker
    {
        private readonly string _id;
        private readonly FlaggedChainedWorker _innerWorker;
        private readonly Action _work;
        private bool _busy;
        private bool _flagged;

        public FlaggedChainedWorker(string id, Action work, FlaggedChainedWorker innerWorker)
        {
            _id = id;
            _work = work;
            _innerWorker = innerWorker;
        }

        public void GetToWork()
        {
            if (_busy)
            {
                _flagged = true;
                return;
            }

            do
            {
                _flagged = false;
                _busy = true;
                Console.WriteLine(String.Format("{0} begin", _id));

                _work.Invoke();

                if (_innerWorker != null)
                {
                    _innerWorker.GetToWork();
                }
                Console.WriteLine(String.Format("{0} end", _id));

                _busy = false;
            } while (_flagged);
        }
    }
}

Answer 3

在我看来，您过于复杂了。我以前写过这些“管道”类；你所需要的只是一个工人队列，每个工人都有一个等待句柄，在操作完成后会收到信号。

public class Pipeline : IDisposable
{
    private readonly IEnumerable<Stage> stages;

    public Pipeline(IEnumerable<Action> actions)
    {
        if (actions == null)
            throw new ArgumentNullException("actions");
        stages = actions.Select(a => new Stage(a)).ToList();
    }

    public Pipeline(params Action[] actions)
        : this(actions as IEnumerable<Action>)
    {
    }

    public void Dispose()
    {
        foreach (Stage stage in stages)
            stage.Dispose();
    }

    public void Start()
    {
        foreach (Stage currentStage in stages)
            currentStage.Execute();
    }

    class Stage : IDisposable
    {
        private readonly Action action;
        private readonly EventWaitHandle readyEvent;

        public Stage(Action action)
        {
            this.action = action;
            this.readyEvent = new AutoResetEvent(true);
        }

        public void Dispose()
        {
            readyEvent.Close();
        }

        public void Execute()
        {
            readyEvent.WaitOne();
            action();
            readyEvent.Set();
        }
    }
}

这是一个测试程序，您可以使用它来验证操作是否始终以正确的顺序执行，并且一次只能执行一个相同的操作：

class Program
{
    static void Main(string[] args)
    {
        Action firstAction = GetTestAction(1);
        Action secondAction = GetTestAction(2);
        Action thirdAction = GetTestAction(3);
        Pipeline pipeline = new Pipeline(firstAction, secondAction, thirdAction);
        for (int i = 0; i < 10; i++)
        {
            ThreadPool.QueueUserWorkItem(s => pipeline.Start());
        }
    }

    static Action GetTestAction(int index)
    {
        return () =>
        {
            Console.WriteLine("Action started: {0}", index);
            Thread.Sleep(100);
            Console.WriteLine("Action finished: {0}", index);
        };
    }
}

简短、简单、完全线程安全。

如果由于某种原因您需要从链中的特定步骤开始工作，那么您只需为Start 添加重载：

public void Start(int index)
{
    foreach (Stage currentStage in stages.Skip(index + 1))
        currentStage.Execute();
}

---

编辑

基于 cmets，我认为内部 Stage 类的一些小改动应该足以获得你想要的那种行为。我们只需要在“ready”事件之外添加一个“queued”事件。

    class Stage : IDisposable
    {
        private readonly Action action;
        private readonly EventWaitHandle readyEvent;
        private readonly EventWaitHandle queuedEvent;

        public Stage(Action action)
        {
            this.action = action;
            this.readyEvent = new AutoResetEvent(true);
            this.queuedEvent = new AutoResetEvent(true);
        }

        public void Dispose()
        {
            readyEvent.Close();
        }

        private bool CanExecute()
        {
            if (readyEvent.WaitOne(0, true))
                return true;
            if (!queuedEvent.WaitOne(0, true))
                return false;
            readyEvent.WaitOne();
            queuedEvent.Set();
            return true;
        }

        public bool Execute()
        {
            if (!CanExecute())
                return false;
            action();
            readyEvent.Set();
            return true;
        }
    }

如果阶段无法执行（即已经排队），还可以更改管道的 Start 方法以中断：

public void Start(int index)
{
    foreach (Stage currentStage in stages.Skip(index + 1))
        if (!currentStage.Execute())
            break;
}

这里的概念很简单，再说一遍：

• 阶段首先尝试立即获取就绪状态。如果成功，则开始运行。
• 如果它无法获取就绪状态（即任务已经在运行），那么它会尝试获取排队状态。
  • 如果它获得排队状态，则等待就绪状态变为可用，然后释放排队状态。
  • 如果也无法获得排队状态，则放弃。

我再次阅读了您的问题和 cmets，我很确定这正是您想要做的，并且在安全性、吞吐量和节流之间进行了最佳权衡。

因为ThreadPool 有时可能需要一段时间才能响应，所以如果您想真正看到“跳过”发生，您应该将测试程序中的延迟提高到1000 而不是100。