转载一篇老外的消息队列服务文章

太过爱你忘了你带给我的痛 2022-08-13 12:44 166阅读 0赞

接着上一篇：[http://blog.csdn.net/joyhen/article/details/38534569][http_blog.csdn.net_joyhen_article_details_38534569]

原文：[http://www.codeproject.com/Articles/821220/Throwing-a-Great-Block][http_www.codeproject.com_Articles_821220_Throwing-a-Great-Block]

using System;
    
    namespace MessagingQueueing
    {
        using System.Collections.Concurrent;
        using System.Threading;
    
        class Program
        {
            static void Main(string[] args)
            {
                using (MyQueue queue = new MyQueue())
                {
                    for (int msgIdx = 1; msgIdx < 21; msgIdx++)
                    {
                        queue.AddLog(new MyMessage
                        {
                            MessageId = msgIdx,
                            Message = string.Format("Message text # {0:#,##0}", msgIdx)
                        });
                    }
                }
    
                Console.ReadKey();
            }
        }
    
        class MyQueue : IDisposable
        {
            private BlockingCollection<MyMessage> messageQueue;
            private Thread dequeueThread;
    
            bool stopped = true;
            bool isStopping = false;
    
            public MyQueue()
            {
                messageQueue = new BlockingCollection<MyMessage>(new ConcurrentQueue<MyMessage>());
                dequeueThread = new Thread(new ThreadStart(DequeueMessageThread));
                dequeueThread.Name = "TransactionPostThread";
                dequeueThread.Start();
                stopped = false;
            }
    
            private void DequeueMessageThread()
            {
                try
                {
                    while (true)
                    {
                        MyMessage message = messageQueue.Take();
                        Console.WriteLine("Dequeueing: " + message.ToString());
    
                        if (messageQueue.IsCompleted)
                        {
                            break;
                        }
                    }
                }
                catch (InvalidOperationException)
                {
                    // if invalid op it's because queue was completed
                }
                catch (ThreadAbortException)
                {
                    // Thread aborted due to queue issue, ignore
                }
                catch (Exception)
                {
                    throw;
                }
            }
    
            public void AddLog(MyMessage message)
            {
                Console.WriteLine("Enqueueing: " + message.ToString());
                messageQueue.Add(message);
            }
    
            /// <summary>
            /// 实现IDisposable接口方法
            /// </summary>
            public void Dispose()
            {
                Dispose(false);
            }
    
            private void Dispose(bool fromDestructor)
            {
                isStopping = true;
                int logShutdownTimeout = 30000;
    
                Console.WriteLine("Shutting down queue. Waiting for dequeue thread completion.");
    
                // Signal queue that we're shutting down
                messageQueue.CompleteAdding();
    
                // Wait for thread to complete before exiting
                do
                {
                    if (!dequeueThread.Join(logShutdownTimeout))
                    {
                        // Queue thread may be stuck. Check for items in queue and kill thread if empty
    
                        if (messageQueue.Count == 0)
                        {
                            System.Diagnostics.Debug.Print("Aborting thread");
                            dequeueThread.Abort();
                            break;
                        }
                    }
                } while (dequeueThread.IsAlive);
    
                Console.WriteLine("Dequeue thread complete.");
    
                if (!fromDestructor)
                {
                    GC.SuppressFinalize(this);
                }
    
                stopped = true;
                isStopping = false;
            }
    
            ~MyQueue()
            {
                Dispose(true);
            }
        }
    
        class MyMessage
        {
            public int MessageId { get; set; }
            public string Message { get; set; }
    
            public override string ToString()
            {
                return string.Format("Message with ID {0:#,##0} and value {1}.", MessageId, Message);
            }
    
        }
    
    }

![SouthEast][]

[http_blog.csdn.net_joyhen_article_details_38534569]: http://blog.csdn.net/joyhen/article/details/38534569
[http_www.codeproject.com_Articles_821220_Throwing-a-Great-Block]: http://www.codeproject.com/Articles/821220/Throwing-a-Great-Block
[SouthEast]: /images/20220810/9dad1af1eba2436b9b2fde9afecf96c3.png