c# - 一种使用 ThreadPool 跟踪失败工作人员的可靠方法

Question

我正在寻找一种很好的方法来跟踪(计数)哪些工作人员在使用线程池排队并使用 WaitHandle.WaitAll() 让所有线程完成时失败。

Interlocking 计数器是一种好的技术还是有更强大的策略？

Answer 1

好的，这是您可以采用的一种方法。我已经将我们要跟踪的数据封装到类 TrackedWorkers 中。此类上有一个构造函数，可让您设置将要工作的 worker 数。然后，使用 LaunchWorkers 启动 worker，这需要一个委托(delegate)吃掉一个 object 并返回一个 bool。 object 代表工作人员的输入，bool 代表成功或失败取决于 true 或 false 是分别返回值。

所以基本上我们所做的是我们有一个数组来跟踪工作人员状态。我们启动工作人员并根据工作人员的返回值设置与该工作人员相对应的状态。当工作人员返回时，我们为所有要设置的 AutoResetEvents 设置一个 AutoResetEvent 和 WaitHandle.WaitAll。

请注意，有一个嵌套类来跟踪工作人员应该完成的工作(委托(delegate))、该工作的输入，以及用于设置状态 AutoResetEvent 的 ID 对应于那个线程。

非常仔细地注意，一旦工作完成，我们就不会持有对工作委托(delegate) func 或 input 的引用。这很重要，这样我们就不会意外地阻止东西被垃圾收集。

有一些方法可以获取特定 worker 的状态，以及所有成功的 worker 的索引和所有失败的 worker 的索引。

最后一点:我不认为此代码生产准备就绪。这只是我将采用的方法的草图。您需要注意添加测试、异常处理和其他此类细节。

class TrackedWorkers {
    class WorkerState {
        public object Input { get; private set; }
        public int ID { get; private set; }
        public Func<object, bool> Func { get; private set; }
        public WorkerState(Func<object, bool> func, object input, int id) {
            Func = func;
            Input = input;
            ID = id;
        }
    }

    AutoResetEvent[] events;
    bool[] statuses;
    bool _workComplete;
    int _number;

    public TrackedWorkers(int number) {
        if (number <= 0 || number > 64) {
            throw new ArgumentOutOfRangeException(
                "number",
                "number must be positive and at most 64"
            );
        }
        this._number = number;
        events = new AutoResetEvent[number];
        statuses = new bool[number];
        _workComplete = false;
    }

    void Initialize() {
        _workComplete = false;
        for (int i = 0; i < _number; i++) {
            events[i] = new AutoResetEvent(false);
            statuses[i] = true;
        }
    }

    void DoWork(object state) {
        WorkerState ws = (WorkerState)state;
        statuses[ws.ID] = ws.Func(ws.Input);
        events[ws.ID].Set();
    }

    public void LaunchWorkers(Func<object, bool> func, object[] inputs) {
        Initialize();
        for (int i = 0; i < _number; i++) {
            WorkerState ws = new WorkerState(func, inputs[i], i);
            ThreadPool.QueueUserWorkItem(this.DoWork, ws);
        }
        WaitHandle.WaitAll(events);
        _workComplete = true;
    }

    void ThrowIfWorkIsNotDone() {
        if (!_workComplete) {
            throw new InvalidOperationException("work not complete");
        }
    }

    public bool GetWorkerStatus(int i) {
        ThrowIfWorkIsNotDone();
        return statuses[i];
    }

    public IEnumerable<int> SuccessfulWorkers {
        get {
            return WorkersWhere(b => b);
        }
    }

    public IEnumerable<int> FailedWorkers {
        get {
            return WorkersWhere(b => !b);
        }
    }

    IEnumerable<int> WorkersWhere(Predicate<bool> predicate) {
        ThrowIfWorkIsNotDone();
        for (int i = 0; i < _number; i++) {
            if (predicate(statuses[i])) {
                yield return i;
            }
        }
    }
}

示例用法:

class Program {
    static Random rg = new Random();
    static object lockObject = new object();
    static void Main(string[] args) {
        int count = 64;
        Pair[] pairs = new Pair[count];
        for(int i = 0; i < count; i++) {
            pairs[i] = new Pair(i, 2 * i);
        }
        TrackedWorkers workers = new TrackedWorkers(count);
        workers.LaunchWorkers(SleepAndAdd, pairs.Cast<object>().ToArray());
        Console.WriteLine(
            "Number successful: {0}",
            workers.SuccessfulWorkers.Count()
        );
        Console.WriteLine(
            "Number failed: {0}",
            workers.FailedWorkers.Count()
        );
    }
    static bool SleepAndAdd(object o) {
        Pair pair = (Pair)o;
        int timeout;
        double d;
        lock (lockObject) {
            timeout = rg.Next(1000);
            d = rg.NextDouble();
        }
        Thread.Sleep(timeout);
        bool success = d < 0.5;
        if (success) {
            Console.WriteLine(pair.First + pair.Second);
        }
        return (success);

    }
}

上面的程序将启动 64 个线程。第 i 线程的任务是将数字 i 和 2 * i 相加并将结果打印到控制台。但是，我添加了一个随机的 sleep 时间(少于一秒)来模拟忙碌，并且我掷硬币来确定线程的成功或失败。那些成功的打印出他们被分配的总和并返回 true。那些失败的不打印任何内容并返回 false。

这里我用过

struct Pair {
    public int First { get; private set; }
    public int Second { get; private set; }
    public Pair(int first, int second) : this() {
        this.First = first;
        this.Second = second;
    }
}