详解异步多线程使用中的常见问题

x33g5p2x  于2022-01-04 转载在 其他  
字(9.7k)|赞(0)|评价(0)|浏览(306)

上一篇:异步多线程之Parallel

异常处理

小伙伴有没有想过,多线程的异常怎么处理,同步方法内的异常处理,想必都非常非常熟悉了。那多线程是什么样的呢,接着我讲解多线程的异常处理

首先,我们定义个任务列表,当 11、12 次的时候,抛出一个异常,最外围使用 try catch 包一下

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    try
    {
        TaskFactory taskFactory = new TaskFactory();
        List<Task> tasks = new List<Task>();
        for (int i = 0; i < 20; i++)
        {
            string name = $"第 {i} 次";

            Action<object> action = t =>
            {
                Thread.Sleep(2 * 1000);
                if (name.ToString().Equals("第 11 次"))
                {
                    throw new Exception($"{t},执行失败");
                }
                if (name.ToString().Equals("第 12 次"))
                {
                    throw new Exception($"{t},执行失败");
                }
                Console.WriteLine($"{t},执行成功");
            };

            tasks.Add(taskFactory.StartNew(action, name));
        }
    }
    catch (AggregateException aex)
    {
        foreach (var item in aex.InnerExceptions)
        {
            Console.WriteLine("Main AggregateException:" + item.Message);
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine("Main Exception:" + ex.Message);
    }

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,可以看到 vs 捕获到了异常的代码行,但 catch 并未捕获到异常,这是为什么呢?是因为线程里面的异常被吞掉了,从运行的结果也可以看到,main end 在子线程没有执行任时就已经结束了,那说明 catch 已经执行过去了。


那有没有办法捕获多线程的异常呢?答案:有的,等待线程完成计算即可

看下面代码,有个特殊的地方 AggregateException.InnerExceptions 专门为多线程准备的,可以查看多线程异常信息

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    try
    {
        TaskFactory taskFactory = new TaskFactory();
        List<Task> tasks = new List<Task>();
        for (int i = 0; i < 20; i++)
        {
            string name = $"第 {i} 次";

            Action<object> action = t =>
            {
                Thread.Sleep(2 * 1000);
                if (name.ToString().Equals("第 11 次"))
                {
                    throw new Exception($"{t},执行失败");
                }
                if (name.ToString().Equals("第 12 次"))
                {
                    throw new Exception($"{t},执行失败");
                }
                Console.WriteLine($"{t},执行成功");
            };

            tasks.Add(taskFactory.StartNew(action, name));
        }

        Task.WaitAll(tasks.ToArray());
    }
    catch (AggregateException aex)
    {
        foreach (var item in aex.InnerExceptions)
        {
            Console.WriteLine("Main AggregateException:" + item.Message);
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine("Main Exception:" + ex.Message);
    }

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动线程,可以看到任务全部执行完毕,且 AggregateException.InnerExceptions 存储了,子线程执行时的异常信息

但 WaitAll 不好,总不能一直 WaitAll 吧,它会卡界面。并不适用于异步场景对吧,接着来看另外一直解决方案。就是子线程里不允许出现异常,如果有自己处理好,即 try catch 包一下,平时工作中建议这么做。

使用 try catch 将子线程执行的代码包一下,且在 catch 打印错误信息

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    try
    {
        TaskFactory taskFactory = new TaskFactory();
        List<Task> tasks = new List<Task>();
        for (int i = 0; i < 20; i++)
        {
            string name = $"第 {i} 次";

            Action<object> action = t =>
            {
                try
                {
                    Thread.Sleep(2 * 1000);
                    if (name.ToString().Equals("第 11 次"))
                    {
                        throw new Exception($"{t},执行失败");
                    }
                    if (name.ToString().Equals("第 12 次"))
                    {
                        throw new Exception($"{t},执行失败");
                    }
                    Console.WriteLine($"{t},执行成功");
                }
                catch (Exception ex)
                {
					Console.WriteLine(ex.Message);
                }
            };

            tasks.Add(taskFactory.StartNew(action, name));
        }
    }
    catch (AggregateException aex)
    {
        foreach (var item in aex.InnerExceptions)
        {
            Console.WriteLine("Main AggregateException:" + item.Message);
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine("Main Exception:" + ex.Message);
    }

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,可以看到任务全部执行,且子线程异常也捕获到

线程取消

有时候会有这样的场景,多个任务并发执行,如果某个任务失败了,通知其他的任务都停下来。首先打个预防针 Task 在外部无法中止的,Thread.Abort 不靠谱。其实线程取消的这个想法是错误的,线程是 OS 的资源,程序是无法掌控什么时候取消,发出一个动作可能立马取消,也可能等 1 s 取消。
解决方案:线程自己停止自己,定义公共的变量,修改变量状态,其他线程不断检测公共变量

例如:CancellationTokenSource 就是公共变量,初始化为 false 状态,程序执行 CancellationTokenSource .Cancel() 方法会取消,其他线程检测到 CancellationTokenSource .IsCancellationRequested 会是取消状态。CancellationTokenSource.Token 在启动 Task 时传入,如果已经 CancellationTokenSource.Cancel() ,这个任务会放弃启动,抛出一个异常的形式放弃。

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    try
    {
        TaskFactory taskFactory = new TaskFactory();
        List<Task> tasks = new List<Task>();
        CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(); // bool 

        for (int i = 0; i < 20; i++)
        {
            string name = $"第 {i} 次";

            Action<object> action = t =>
            {
                try
                {
                    Thread.Sleep(2 * 1000);
                    if (name.ToString().Equals("第 11 次"))
                    {
                        throw new Exception($"{t},执行失败");
                    }
                    if (name.ToString().Equals("第 12 次"))
                    {
                        throw new Exception($"{t},执行失败");
                    }
                    if (cancellationTokenSource.IsCancellationRequested) // 检测信号量
                    {
                        Console.WriteLine($"{t},放弃执行");
                        return;
                    }
                    Console.WriteLine($"{t},执行成功");
                }
                catch (Exception ex)
                {
                    cancellationTokenSource.Cancel();
                    Console.WriteLine(ex.Message);
                }
            };

            tasks.Add(taskFactory.StartNew(action, name,cancellationTokenSource.Token));
        }

        Task.WaitAll(tasks.ToArray());
    }
    catch (AggregateException aex)
    {
        foreach (var item in aex.InnerExceptions)
        {
            Console.WriteLine("Main AggregateException:" + item.Message);
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine("Main Exception:" + ex.Message);
    }

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,可以看到 11、12 此任务失败,18、19 放弃了任务执。有的小伙伴疑问了,12 之后的部分为什么执行成功了,因为 CPU 是分时分片的吗,会有延迟,延迟少不了。

临时变量

首先看个代码,循环 5 次,多线程的方式,依次输出序号

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    for (int i = 0; i < 5; i++)
    {
        Task.Run(() => {
            Console.WriteLine(i);
        });
    }

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,不是我们预期的结果 0、1、2、3、4,为什么是 5 个 5 呢?因为全程只有一个 i ,当主线程执行完毕时 i = 5 ,但子线程可能还没有开始执行任务,轮到子线程取 i 时,已经是主线程 1 循环完毕后的 5 了。

改造代码:在 for 循环内加一行代码 int k = i,且在子线程用的变量也改为 k

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    for (int i = 0; i < 5; i++)
    {
        int k = i;
        Task.Run(() => {
            Console.WriteLine($"k={k},i={i}");
        });
    }

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,可以看到是我们预期的结果 0、1、2、3、4,为什么会这样子呢?因为全程有 5 个 k,每次循环都会创建一个 k 存储当前的 i,不同的子线程使用的也是,每次循环的 i 值。

线程安全

首先为什么会有线程安全的概念呢?首先我们来看一个正常程序,如下

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    int TotalCount = 0;
    List<int> vs = new List<int>();

    for (int i = 0; i < 10000; i++)
    {
        TotalCount += 1;
        vs.Add(i);
    }

    Console.WriteLine(TotalCount);
    Console.WriteLine(vs.Count);

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,可以看到循环 10000 次,最终的求和与列表里的数据量都是 10000,这是正常的


接着,将求和与添加列表,换成多线程,等待全部线程完成工作后,打印信息

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    int TotalCount = 0;
    List<int> vs = new List<int>();

    TaskFactory taskFactory = new TaskFactory();
    List<Task> tasks = new List<Task>();

    for (int i = 0; i < 10000; i++)
    {
        int k = i;
        tasks.Add(taskFactory.StartNew(() =>
        {
            TotalCount += 1;
            vs.Add(i);
        }));
    }

    Task.WaitAll(tasks.ToArray());

    Console.WriteLine(TotalCount);
    Console.WriteLine(vs.Count);

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

启动程序,可以看到,两个结果都不是 10000 呢?这就是线程安全

因为 TotalCount 是个共享的变量,当多个线程去取 TotalCount 进行 +1 后,线程都去放值的时候,后一个线程会替换掉前一个线程放置的值,所以就会形成做最终不是 10000 的结果。列表,可以看做是一个连续的块,当多线程添加的时候,也会进行覆盖。

如何解决呢?答案:lock、安全队列、拆分合并计算。下面对 lock 进行讲解,安全队列与拆分合并计算,有兴趣的小伙伴可以私下交流

1 .lock
第一种,通过加锁的方式,这种也是日常工作总常用的一种。首先定义个私有的静态引用类型的变量,然后将需要锁的运算放到 lock () 方法内

在 { } 内同一时刻,只有一个线程执行,所以尽可能 {} 放置必要的逻辑运行提高效率。lock 只能锁引用类型,原理是占用这个引用链接。不要用 string 会享元,即如 lock() 是相同的字符串,无论定义多少个变量,其实都是一个。

internal class Program
{
    private static readonly object _lock = new object();

    static void Main(string[] args)
    {
        Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

        int TotalCount = 0;
        List<int> vs = new List<int>();

        TaskFactory taskFactory = new TaskFactory();
        List<Task> tasks = new List<Task>();

        for (int i = 0; i < 10000; i++)
        {
            int k = i;
            tasks.Add(taskFactory.StartNew(() =>
            {
                lock (_lock)
                {
                    TotalCount += 1;
                    vs.Add(i);
                }
            }));
        }

        Task.WaitAll(tasks.ToArray());

        Console.WriteLine(TotalCount);
        Console.WriteLine(vs.Count);

        Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

        Console.ReadLine();
    }
}

启动程序,可以看到,此时在多线程的情况下,最终的结果是正常的

这段代码,是官方推荐写法 private 防止外面也被引用,static 保证全场唯一

private static readonly object _lock = new object();

扩展:与 lock 等价的有个 Monitor,用法如下

private static object _lock = new object();

static void Main(string[] args)
{
    Console.WriteLine($"Main Start,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    int TotalCount = 0;
    List<int> vs = new List<int>();

    TaskFactory taskFactory = new TaskFactory();
    List<Task> tasks = new List<Task>();

    for (int i = 0; i < 10000; i++)
    {
        int k = i;
        tasks.Add(taskFactory.StartNew(() =>
        {
            Monitor.Enter(_lock);
            TotalCount += 1;
            vs.Add(i);
            Monitor.Exit(_lock);
        }));
    }

    Task.WaitAll(tasks.ToArray());

    Console.WriteLine(TotalCount);
    Console.WriteLine(vs.Count);

    Console.WriteLine($"Main End,ThreadId:{Thread.CurrentThread.ManagedThreadId},Datetime:{DateTime.Now.ToLongTimeString()}");

    Console.ReadLine();
}

相关文章