如何处理使用 SemaphoreSlim 时挂起的线程

Question

我有一些通过第三方库运行数千个 URL 的代码。有时库中的方法会挂起，这会占用一个线程。过了一会儿，所有线程都被什么都不做的进程占用了，然后它就停止了。

我使用SemaphoreSlim 来控制添加新线程，这样我就可以运行最佳数量的任务。我需要一种方法来识别运行时间过长的任务，然后杀死它们，同时从SemaphoreSlim 释放一个线程，以便可以创建一个新任务。

我正在为这里的方法苦苦挣扎，所以我制作了一些模仿我正在做的事情的测试代码。它创建的任务有 10% 的机会挂起，很快所有线程都挂起。

我应该如何检查并杀死它们？

代码如下：

class Program
{
    public static SemaphoreSlim semaphore;
    public static List<Task> taskList;
    static void Main(string[] args)
    {

        List<string> urlList = new List<string>();
        Console.WriteLine("Generating list");
        for (int i = 0; i < 1000; i++)
        {
            //adding random strings to simulate a large list of URLs to process
            urlList.Add(Path.GetRandomFileName());
        }
        Console.WriteLine("Queueing tasks");

        semaphore = new SemaphoreSlim(10, 10);

        Task.Run(() => QueueTasks(urlList));

        Console.ReadLine();
    }
    static void QueueTasks(List<string> urlList)
    {
        taskList = new List<Task>();

        foreach (var url in urlList)
        {
            Console.WriteLine("{0} tasks can enter the semaphore.",
                  semaphore.CurrentCount);
            semaphore.Wait();

            taskList.Add(DoTheThing(url));
        }
    }
    static async Task DoTheThing(string url)
    {

        Random rand = new Random();

        // simulate the IO process
        await Task.Delay(rand.Next(2000, 10000));

        // add a 10% chance that the thread will hang simulating what happens occasionally with http request
        int chance = rand.Next(1, 100);
        if (chance <= 10)
        {
            while (true)
            {
                await Task.Delay(1000000);
            }
        }

        semaphore.Release();
        Console.WriteLine(url);
    }
}

Answer 1

正如人们已经指出的那样，中止线程通常是不好的，并且在 C# 中没有保证这样做的方法。使用一个单独的进程来完成工作，然后杀死它比尝试 Thread.Abort 稍微好一点；但仍然不是最好的方法。理想情况下，您需要协作线程/进程，它们使用 IPC 来决定何时自行解救。这样清理就可以正确完成了。

话虽如此，您可以使用下面的代码来执行您打算执行的操作。我已经写了它假设你的任务将在一个线程中完成。稍作改动，您就可以使用相同的逻辑在一个流程中完成您的任务

代码绝不是万无一失的，只是为了说明。并发代码没有经过真正的测试。锁的持有时间超过了需要，有些地方我没有锁定（比如 Log 函数）

class TaskInfo {
    public Thread Task;
    public DateTime StartTime;

    public TaskInfo(ParameterizedThreadStart startInfo, object startArg) {
        Task = new Thread(startInfo);
        Task.Start(startArg);
        StartTime = DateTime.Now;
    }

}

class Program {

    const int MAX_THREADS = 1;
    const int TASK_TIMEOUT = 6; // in seconds
    const int CLEANUP_INTERVAL = TASK_TIMEOUT; // in seconds

    public static SemaphoreSlim semaphore;

    public static List<TaskInfo> TaskList;
    public static object TaskListLock = new object();

    public static Timer CleanupTimer;

    static void Main(string[] args) {
        List<string> urlList = new List<string>();
        Log("Generating list");
        for (int i = 0; i < 2; i++) {
            //adding random strings to simulate a large list of URLs to process
            urlList.Add(Path.GetRandomFileName());
        }
        Log("Queueing tasks");

        semaphore = new SemaphoreSlim(MAX_THREADS, MAX_THREADS);

        Task.Run(() => QueueTasks(urlList));

        CleanupTimer = new Timer(CleanupTasks, null, CLEANUP_INTERVAL * 1000, CLEANUP_INTERVAL * 1000);


        Console.ReadLine();
    }

    // TODO: Guard against re-entrancy
    static void CleanupTasks(object state) {
        Log("CleanupTasks started");

        lock (TaskListLock) {
            var now = DateTime.Now;
            int n = TaskList.Count;
            for (int i = n - 1; i >= 0; --i) {
                var task = TaskList[i];
                Log($"Checking task with ID {task.Task.ManagedThreadId}");

                // kill processes running for longer than anticipated
                if (task.Task.IsAlive && now.Subtract(task.StartTime).TotalSeconds >= TASK_TIMEOUT) {
                    Log("Cleaning up hung task");
                    task.Task.Abort();
                }

                // remove task if it is not alive
                if (!task.Task.IsAlive) {
                    Log("Removing dead task from list");
                    TaskList.RemoveAt(i);
                    continue;
                }

            }

            if (TaskList.Count == 0) {
                Log("Disposing cleanup thread");
                CleanupTimer.Dispose();
            }
        }

        Log("CleanupTasks done");
    }

    static void QueueTasks(List<string> urlList) {
        TaskList = new List<TaskInfo>();

        foreach (var url in urlList) {
            Log($"Trying to schedule url = {url}");
            semaphore.Wait();
            Log("Semaphore acquired");

            ParameterizedThreadStart taskRoutine = obj => {
                try {
                    DoTheThing((string)obj);
                } finally {
                    Log("Releasing semaphore");
                    semaphore.Release();
                }
            };

            var task = new TaskInfo(taskRoutine, url);
            lock (TaskListLock)
                TaskList.Add(task);
        }

        Log("All tasks queued");
    }

    // simulate all processes get hung
    static void DoTheThing(string url) {
        while (true)
            Thread.Sleep(5000);
    }

    static void Log(string msg) {
        Console.WriteLine("{0:HH:mm:ss.fff} Thread {1,2} {2}", DateTime.Now, Thread.CurrentThread.ManagedThreadId.ToString(), msg);
    }
}