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