关于并发下内存及CPU使用情况的思考

Posted 2020-10-22 又见阿郎

鉴于昨天的文章<<使用Interlocked在多线程下进行原子操作，无锁无阻塞的实现线程运行状态判断>>里面有一个封装好的无锁的类库可以判断并发下的结束状况，我们可以完成并发时，以及并发的同时做一些事，因此，今天我做了个小demo:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApp6
{
    
    public enum CoordinationStatus
    {
        AllDone,
        Timeout,
        Cancel
    };

    public sealed class AsyncCoordinator
    {
        private Int32 m_opCount = 1;        // Decremented when AllBegun calls JustEnded
        private Int32 m_statusReported = 0; // 0=false, 1=true
        private Action<CoordinationStatus> m_callback;
        private Timer m_timer;

        // This method MUST be called BEFORE initiating an operation
        public void AboutToBegin(Int32 opsToAdd = 1)
        {
            Interlocked.Add(ref m_opCount, opsToAdd);
        }

        // This method MUST be called AFTER an operations result has been processed
        public void JustEnded()
        {
            if (Interlocked.Decrement(ref m_opCount) == 0)
                ReportStatus(CoordinationStatus.AllDone);
        }

        // This method MUST be called AFTER initiating ALL operations
        public void AllBegun(Action<CoordinationStatus> callback, Int32 timeout = Timeout.Infinite)
        {
            m_callback = callback;
            // 是否需要永远运行
            if (timeout != Timeout.Infinite)
            {
                // 在指定的时间点(dueTime) 调用回调函数，随后在指定的时间间隔(period)调用回调函数
                m_timer = new Timer(TimeExpired, null, timeout, Timeout.Infinite);
            }
            JustEnded();
        }

        // 处理过时的线程
        private void TimeExpired(Object o)
        {
            ReportStatus(CoordinationStatus.Timeout);
        }

        public void Cancel()
        {
            if (m_callback == null)
                throw new InvalidOperationException("Cancel cannot be called before AllBegun");
            ReportStatus(CoordinationStatus.Cancel);
        }

        private void ReportStatus(CoordinationStatus status)
        {
            if (m_timer != null)
            {  // If timer is still in play, kill it
                Timer timer = Interlocked.Exchange(ref m_timer, null);
                if (timer != null) timer.Dispose();
            }

            // If status has never been reported, report it; else ignore it
            if (Interlocked.Exchange(ref m_statusReported, 1) == 0)
                m_callback(status);
        }
    }

    

    class Program
    {
        static AsyncCoordinator m_ac = new AsyncCoordinator();
        static void Main(string[] args)
        {
            Console.BufferHeight = Int16.MaxValue - 10;
            Console.BufferWidth = Int16.MaxValue - 10;

            ConcurrentQueue<int> concurrentQueue = new ConcurrentQueue<int>();

            for(int i =0;i<10000; i++)
            {
                concurrentQueue.Enqueue(i);
            }

            Console.WriteLine("添加完毕....");

            var t = new Task[50];
            
            for (int i=0; i<50; i++)
            {
                m_ac.AboutToBegin(1);
                t[i] = Task.Factory.StartNew((param) =>
                {
                    while (concurrentQueue.Count>0)
                    {
                        int x;
                        if(concurrentQueue.TryDequeue(out x))
                        {
                            Console.WriteLine(x + "   线程Id: {0},   线程数: {1}", Task.CurrentId, param.ToString());
                        }
                        //Thread.Sleep(150);
                    }
                    m_ac.JustEnded();
                }, i);
            }
            
            m_ac.AllBegun(AllDone, Timeout.Infinite);
            Console.ReadKey();
        }

        public static void AllDone(CoordinationStatus status)
        {
            switch (status)
            {
                case CoordinationStatus.Cancel:
                    Console.WriteLine("Operation canceled.");
                    break;

                case CoordinationStatus.Timeout:
                    Console.WriteLine("Operation timed-out.");
                    break;

                case CoordinationStatus.AllDone:
                    Console.WriteLine("处理完毕....");
                    break;
            }
        }
    }
}

但是发现了一个问题:

这CPU使用率....
然后我看了下输出结果:

可以看到线程数才只有5个(我的线程数是从0开始算的)，这不会啊，明明我们就开了50个线程啊，不过不管开多少个线程，这CPU扛不住啊，要是说在项目中的某个模块需要用到并发，这CPU使用率你扛得住？服务器本来配置就不会太好，网站的其余模块不要用CPU了？而且，我明明开了50个线程跑啊，为什么只有五个线程？其实很简单，因此并发下，代码只用了五个线程就跑完了这一万个数据，剩下的线程开了没有用武之地。找到只有五个线程开着的原因了之后，要想想怎么解决啊，多的45个线程也是要占内存的，尽管是线程池线程，但也是要占用内存啊，既然是因为并发下运行太快，只要五个线程就能跑满一万个数据，那我就阻塞一会线程就可以了，这样让剩下的45个线程能够有机会运行。改代码！
将上面的Thread.Sleep(150)的注释给去掉哦！

嗯，这个结果还是可以的，但是有个Console host占用内存高啊，占就占呗，反正该用的内存还是要用。我们睡眠了一段时间的线程，那么与不睡眠相比，并发的CPU使用率是不是下降了？我们开线程最好的期待不就是跑满CPU么？其实不然，开线程不过就是为了更快的运行程序，将耗时的程序分批次运行，但是如果期间占用CPU太高，我这里是个demo，占用CPU时间很短，也就几十秒。但是真的项目中会允许么？具体情况具体分析吧，如果不介意的话，可以这么跑，大不了另外弄个服务器专门跑并发，然后将数据存储到数据库中(如果你的业务是: 并发调用第三方接口，然后将接口获取的数据做处理，完全可以采用这种设计)。但是请注意，还是不要太耗费CPU的好。
并发线程的睡眠时间，我们也可以自己调节下，建议是100-200ms吧。