c# 安全队列
2016-09-22 21:05
176 查看
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace AA
{
public class AsynQueue<T>
{
//队列是否正在处理数据
private int isProcessing;
//有线程正在处理数据
private const int Processing = 1;
//没有线程处理数据
private const int UnProcessing = 0;
//队列是否可用
private volatile bool enabled = true;
private Task currentTask;
public event Action<T> ProcessItemFunction;
public event EventHandler<EventArgs<Exception>> ProcessException;
private ConcurrentQueue<T> queue;
public AsynQueue()
{
queue = new ConcurrentQueue<T>();
Start();
}
public int Count
{
get
{
return queue.Count;
}
}
private void Start()
{
Thread process_Thread = new Thread(PorcessItem);
process_Thread.IsBackground = true;
process_Thread.Start();
}
public void Enqueue(T items)
{
if (items == null)
{
throw new ArgumentException("items");
}
queue.Enqueue(items);
DataAdded();
}
//数据添加完成后通知消费者线程处理
private void DataAdded()
{
if (enabled)
{
if (!IsProcessingItem())
{
currentTask = Task.Factory.StartNew(ProcessItemLoop);
}
}
}
//判断是否队列有线程正在处理
private bool IsProcessingItem()
{
return !(Interlocked.CompareExchange(ref isProcessing, Processing, UnProcessing) == 0);
}
private void ProcessItemLoop()
{
if (!enabled && queue.IsEmpty)
{
Interlocked.Exchange(ref isProcessing, 0);
return;
}
T publishFrame;
if (queue.TryDequeue(out publishFrame))
{
try
{
ProcessItemFunction(publishFrame);
}
catch (Exception ex)
{
OnProcessException(ex);
}
}
if (enabled && !queue.IsEmpty)
{
currentTask = Task.Factory.StartNew(ProcessItemLoop);
}
else
{
Interlocked.Exchange(ref isProcessing, UnProcessing);
}
}
/// <summary>
///定时处理线程调用函数
///主要是监视入队的时候线程 没有来的及处理的情况
/// </summary>
private void PorcessItem(object state)
{
int sleepCount = 0;
int sleepTime = 1000;
while (enabled)
{
//如果队列为空则根据循环的次数确定睡眠的时间
if (queue.IsEmpty)
{
if (sleepCount == 0)
{
sleepTime = 1000;
}
else if (sleepCount <= 3)
{
sleepTime = 1000 * 3;
}
else
{
sleepTime = 1000 * 50;
}
sleepCount++;
Thread.Sleep(sleepTime);
}
else
{
//判断是否队列有线程正在处理
if (enabled && Interlocked.CompareExchange(ref isProcessing, Processing, UnProcessing) == 0)
{
if (!queue.IsEmpty)
{
currentTask = Task.Factory.StartNew(ProcessItemLoop);
}
else
{
Interlocked.Exchange(ref isProcessing, 0);
}
sleepCount = 0;
sleepTime = 1000;
}
}
}
}
public void Flsuh()
{
Stop();
if (currentTask != null)
{
currentTask.Wait();
}
while (!queue.IsEmpty)
{
try
{
T publishFrame;
if (queue.TryDequeue(out publishFrame))
{
ProcessItemFunction(publishFrame);
}
}
catch (Exception ex)
{
OnProcessException(ex);
}
}
currentTask = null;
}
public void Stop()
{
this.enabled = false;
}
private void OnProcessException(System.Exception ex)
{
var tempException = ProcessException;
Interlocked.CompareExchange(ref ProcessException, null, null);
if (tempException != null)
{
ProcessException(ex, new EventArgs<Exception>(ex));
}
}
[Serializable]
public class EventArgs<T> : System.EventArgs
{
public T Argument;
public EventArgs() : this(default(T))
{
}
public EventArgs(T argument)
{
Argument = argument;
}
}
}
}
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace AA
{
public class AsynQueue<T>
{
//队列是否正在处理数据
private int isProcessing;
//有线程正在处理数据
private const int Processing = 1;
//没有线程处理数据
private const int UnProcessing = 0;
//队列是否可用
private volatile bool enabled = true;
private Task currentTask;
public event Action<T> ProcessItemFunction;
public event EventHandler<EventArgs<Exception>> ProcessException;
private ConcurrentQueue<T> queue;
public AsynQueue()
{
queue = new ConcurrentQueue<T>();
Start();
}
public int Count
{
get
{
return queue.Count;
}
}
private void Start()
{
Thread process_Thread = new Thread(PorcessItem);
process_Thread.IsBackground = true;
process_Thread.Start();
}
public void Enqueue(T items)
{
if (items == null)
{
throw new ArgumentException("items");
}
queue.Enqueue(items);
DataAdded();
}
//数据添加完成后通知消费者线程处理
private void DataAdded()
{
if (enabled)
{
if (!IsProcessingItem())
{
currentTask = Task.Factory.StartNew(ProcessItemLoop);
}
}
}
//判断是否队列有线程正在处理
private bool IsProcessingItem()
{
return !(Interlocked.CompareExchange(ref isProcessing, Processing, UnProcessing) == 0);
}
private void ProcessItemLoop()
{
if (!enabled && queue.IsEmpty)
{
Interlocked.Exchange(ref isProcessing, 0);
return;
}
T publishFrame;
if (queue.TryDequeue(out publishFrame))
{
try
{
ProcessItemFunction(publishFrame);
}
catch (Exception ex)
{
OnProcessException(ex);
}
}
if (enabled && !queue.IsEmpty)
{
currentTask = Task.Factory.StartNew(ProcessItemLoop);
}
else
{
Interlocked.Exchange(ref isProcessing, UnProcessing);
}
}
/// <summary>
///定时处理线程调用函数
///主要是监视入队的时候线程 没有来的及处理的情况
/// </summary>
private void PorcessItem(object state)
{
int sleepCount = 0;
int sleepTime = 1000;
while (enabled)
{
//如果队列为空则根据循环的次数确定睡眠的时间
if (queue.IsEmpty)
{
if (sleepCount == 0)
{
sleepTime = 1000;
}
else if (sleepCount <= 3)
{
sleepTime = 1000 * 3;
}
else
{
sleepTime = 1000 * 50;
}
sleepCount++;
Thread.Sleep(sleepTime);
}
else
{
//判断是否队列有线程正在处理
if (enabled && Interlocked.CompareExchange(ref isProcessing, Processing, UnProcessing) == 0)
{
if (!queue.IsEmpty)
{
currentTask = Task.Factory.StartNew(ProcessItemLoop);
}
else
{
Interlocked.Exchange(ref isProcessing, 0);
}
sleepCount = 0;
sleepTime = 1000;
}
}
}
}
public void Flsuh()
{
Stop();
if (currentTask != null)
{
currentTask.Wait();
}
while (!queue.IsEmpty)
{
try
{
T publishFrame;
if (queue.TryDequeue(out publishFrame))
{
ProcessItemFunction(publishFrame);
}
}
catch (Exception ex)
{
OnProcessException(ex);
}
}
currentTask = null;
}
public void Stop()
{
this.enabled = false;
}
private void OnProcessException(System.Exception ex)
{
var tempException = ProcessException;
Interlocked.CompareExchange(ref ProcessException, null, null);
if (tempException != null)
{
ProcessException(ex, new EventArgs<Exception>(ex));
}
}
[Serializable]
public class EventArgs<T> : System.EventArgs
{
public T Argument;
public EventArgs() : this(default(T))
{
}
public EventArgs(T argument)
{
Argument = argument;
}
}
}
}
相关文章推荐
- C# 二分法查找和排序
- WPF教程(五) XAML是什么?
- 【C#】无损转换Image为Icon
- C#编程基础(一)
- 关于C#子窗体的Show方法和ShowDialog方法的一些体会
- 多阈值处理利用双Otsu阈值-C#实现-基于EmguCv
- 文字打字机效果
- C# $字符串
- C#面向对象_封装_160922
- C#操作注册表
- C#中Split用法(可以用在取小数点之前后之后)
- winform TabControl 添加关闭按钮 Add close button to TabControl TabPages - C#
- 追着嘟嘟学c#系列基础第十二篇-----c#递归【蓝鸥出品】
- C#程序如何获得系统权限
- Kinect Fusion Basics-WPF C# Sample代码解析
- C#之WinForm基础 画扇形时,出现的从X轴到扇形的第一条边沿顺时针方向的角怎么理解
- C#之WinForm基础 画一个矩形与扇形
- C# XML文件操作类XmlHelper类的完整代码
- c# 添加图片水印,可以指定水印位置+生成缩略图
- C#实现根据IP 查找真实地址