MSMQ的后盾FIFO队列（C＃。NET）
在队列，既高性能，需要时可以处理大量的条目不吹我的系统的内存出水面的要求，我创建了以下实用队列。
总之，它是一个标准的内存中的通用。NET队列，配置溢出达到限制时，将开始使用MSMQ作为后端，以减轻内存使用的问题。在低量的情况下使用时，队列中的行为像一个正常的泛型队列。
任何疑问，意见，随时给我发电子邮件。
另外，看到我在CodeProject上的文章在这里:
。public class OverflowQueue<T> where T : class
{
Queue<T> internalQueue = new Queue<T>();
BinaryMessageFormatter messageFormatter = new BinaryMessageFormatter();
internal static int compactOverflowQueue =
AppConfigHelper.GetValue("CompactOverflowQueueAfterMessagesDequeued", 10000);
internal static int maxInternalQueueSize =
AppConfigHelper.GetValue("MaxInternalOverflowQueueSize", 10000);
QueuePath queuePath;
MessageQueue overflowMSMQ;
long currentQueueSize;
long currentMSMQSize;
bool useBackgroundMsmqPull;
Thread backgroundMsmqPullThread;
volatile bool pushingToMSMQ;
object msmqPullLock = new object();
object msmqLock = new object();
int itemDequeued;
public long CurrentMemoryQueueSize
{
get
{
return currentQueueSize;
}
}
public long CurrentMSMQSize
{
get
{
return Interlocked.Read(ref currentMSMQSize);
}
}
public long TotalCount
{
get
{
return Interlocked.Read(ref currentMSMQSize) +
Interlocked.Read(ref currentQueueSize);
}
}
public OverflowQueue(string queueName, bool purgeQueue, bool useBackgroundMsmqPull)
{
queuePath = new QueuePath(queueName, Environment.MachineName, false);
overflowMSMQ = QueueCreator.CreateAndReturnQueue(queuePath);
overflowMSMQ.Formatter = new BinaryMessageFormatter();
if (purgeQueue)
overflowMSMQ.Purge();
this.useBackgroundMsmqPull = useBackgroundMsmqPull;
if (useBackgroundMsmqPull)
{
// start a thread that pulls msmq messages back into the queue
backgroundMsmqPullThread = new Thread(BackgroundMsmqPull);
backgroundMsmqPullThread.IsBackground = true;
backgroundMsmqPullThread.Name = "BackgroundMsmqPullThread";
backgroundMsmqPullThread.Start();
}
}
void BackgroundMsmqPull()
{
while (true)
{
PullFromMSMQ();
Thread.Sleep(1000);
}
}
public void Enqueue(T item)
{
if (!pushingToMSMQ)
{
if (currentQueueSize >= maxInternalQueueSize)
{
// We've busted the queue size...
// start pushing to MSMQ until the current
// queue zeros out,
// then pop all entries (up to max) back into memory.
lock (msmqLock)
{
pushingToMSMQ = true;
PushToMSMQ(item);
}
}
else
{
// We're still pushing into the memory queue
PushToMemoryQueue(item);
}
}
else
{
// This lock looks like this (split if) because
// I don't wnat to hold onto the lock for a push to the memory queue
lock (msmqLock)
{
if (pushingToMSMQ) // verify we're still pushing to MSMQ
{
PushToMSMQ(item);
return; // Skip the push to memory queue
}
}
// This will only get hit if we aren't still pushing to MSMQ
PushToMemoryQueue(item);
}
}
void PushToMemoryQueue(T item)
{
lock (internalQueue)
internalQueue.Enqueue(item);
Interlocked.Increment(ref currentQueueSize);
}
void PushToMSMQ(T item)
{
Message message = new Message(item, messageFormatter);
overflowMSMQ.Send(message);
Interlocked.Increment(ref currentMSMQSize);
}
public T Dequeue()
{
try
{
if (!useBackgroundMsmqPull)
{
PullFromMSMQ();
}
else
{
// This is here because if the background pull
// is on and a user tries to dequeue too quickly,
// they can run up against an empty memory queue,
// but there could still be something in the MSMQ.
// So, we need to double check for them ;)
if (Interlocked.Read(ref currentQueueSize) == 0
&& Interlocked.Read(ref currentMSMQSize) > 0)
PullFromMSMQ();
}
T item = null;
if (Interlocked.Read(ref currentQueueSize) > 0)
{
lock(internalQueue)
item = internalQueue.Dequeue();
}
if (item != null)
{
Interlocked.Increment(ref itemDequeued);
Interlocked.Decrement(ref currentQueueSize);
return item;
}
else
{
throw new Exception("Nothing to dequeue!");
}
}
finally
{
if (itemDequeued >= compactOverflowQueue)
{
lock (internalQueue)
{
if (itemDequeued >= compactOverflowQueue)
{
// Compact the internal queue to save space
internalQueue.TrimExcess();
itemDequeued = 0;
}
}
}
}
}
void PullFromMSMQ()
{
// We've been putting all new messages into MSMQ...
// now is the time to get them out and put them back into memory.
while (Interlocked.Read(ref currentMSMQSize) > 0 &&
Interlocked.Read(ref currentQueueSize) < maxInternalQueueSize)
// currentQueueSize should be low here
{
Message message = overflowMSMQ.Receive();
// decrement the MSMQ size
Interlocked.Decrement(ref currentMSMQSize);
T item = message.Body as T;
PushToMemoryQueue(item);
}
if (Interlocked.Read(ref currentMSMQSize) <= 0)
{
lock (msmqLock)
// lock around this to prevent the count
// from being wrong when turning off msmq pushing
{
if (Interlocked.Read(ref currentMSMQSize) <= 0)
pushingToMSMQ = false;
}
}
}
}