消息队列实现订单异步提交

what

MSMQ（Microsoft Message Queue），微软消息队列，用于应用程序之间相互通信的一种异步传输模式。应用程序可以分布在同台机器上，也可以分布于互联的网络中的任意位置。基本原理：消息发送者把要发送的消息放入容器，也就是Message（消息），然后保存到系统公用空间的消息队列中（Message Queue）中，本地或互联位置上的消息接收程序再从队列中取出发给它的消息进行处理。消息类型可以是文本，图像，自定义对象等。消息队列分为公共队列和私有队列。

 

why

一、用于进程间的通信。A进程把消息放进消息队列,B进程去消息队列获取，属于异步通信。

二、用于WEB并发请求时作缓冲排队写入。当多个请求并发插入数据库时可能会造成服务器不可用，用户等待时间过长 ，这时可以把所有请求全部放到消息队列中，然后监视进程按先进先出顺序依次写入数据库。

 

how

下面以一个精简版的petshop4.0订单提交代码为例讲解用消息队列实现异步提交的方法步骤。

一、安装MSMQ，在添加删除组件界面勾选安装。

二、新建私有队列OrderQueue，可以安装一个QueueExplorer随时查看消息队列信息。

三、提交订单时如果选择异步提交，则先把订单信息对象插入到消息队列，立即返回信息给用户。

四、服务器运行一个多线程的监视进程不断从消息队列提取订单对象，然后插入到数据库。注意入队列和出队列的对象Formatter一定要相同不然会报异常。

五、部分代码实现如下：

订单实体类：

    [Serializable]
    public class Order
    {
        public string OrderID { get; set; }

        public int MemberID { get; set; }

        public string GoodsID { get; set; }

        public DateTime SubmitTime { get; set; }
    }

基本队列类：

public class BaseQueue: IDisposable {
        protected MessageQueue queue;

        public BaseQueue(string queuePath)
        {
            queue = new MessageQueue(queuePath);
        }

        /// <summary>
        /// 出队列
        /// </summary>
        public virtual object Receive()
        {
            try
            {
                using (Message message = queue.Receive())
                {
                    message.Formatter = queue.Formatter;
                    return message;
                }
            }
            catch (MessageQueueException mqex)
            {
                throw;
            }
        }


        /// <summary>
        /// 入队列
        /// </summary>
        public virtual void Send(object msg)
        {
            queue.Send(msg);
        }


        public void Dispose()
        {
            queue.Dispose();
        }
    }

订单队列类：

  public class OrderQueue : BaseQueue
    {
        // Path example - FormatName:DIRECT=OS:MyMachineNamePrivate$OrderQueueName
        private static readonly string queuePath = ".\private$\orderqueue";

        public OrderQueue()
            : base(queuePath)
        {
            // 可以设置序列化格式
            queue.Formatter = new BinaryMessageFormatter();
        }

        /// <summary>
        /// 从消息队列出一个订单对象，先入先出
        /// </summary>
        public Order Receive()
        {
            object obj = base.Receive();
            return (Order)((Message)obj).Body;
        }

        /// <summary>
        /// 把一个订单对象压入消息队列
        /// </summary>
        public void Send(Order orderMessage)
        {
            base.Send(orderMessage);
        }
    }

提交订单代码：

            Order or = new Order();
            or.OrderID = DateTime.Now.ToString("yyyyMMddHHmmss");
            or.MemberID = int.Parse(TextBox2.Text.Trim());
            or.GoodsID = DropDownList1.SelectedItem.Text.ToString();
            or.SubmitTime = DateTime.Now;

            try
            {
                OrderQueue orQ = new OrderQueue();
                orQ.Send(or);
                Label1.Text = "提交成功！";
            }
            catch (Exception ex)
            {
                Label1.Text = "提交失败！详细原因：" + ex.Message;
            }

多线程读取代码：

       private static int batchSize = 4;//int.Parse(ConfigurationManager.AppSettings["BatchSize"]);
        private static int threadCount = 1;//int.Parse(ConfigurationManager.AppSettings["ThreadCount"]);
        Thread[] workerThreads = new Thread[threadCount];

        private static int totalOrdersProcessed = 0;

        public Form1()
        {
            InitializeComponent();
        }

        //启动线程
        private void button1_Click(object sender, EventArgs e)
        {
            Thread workTicketThread;

            for (int i = 0; i < threadCount; i++)
            {

                workTicketThread = new Thread(new ThreadStart(ProcessOrders));

                // Make this a background thread, so it will terminate when the main thread/process is de-activated
                workTicketThread.IsBackground = true;
                workTicketThread.SetApartmentState(ApartmentState.STA);

                // Start the Work
                workTicketThread.Start();
                workerThreads[i] = workTicketThread;
            }

            MessageBox.Show("starting...");
        }


        //结束线程
        private void button2_Click(object sender, EventArgs e)
        {
            //abort all threads
            for (int i = 0; i < workerThreads.Length; i++)
            {

                workerThreads[i].Abort();
            }
            MessageBox.Show(totalOrdersProcessed + " Orders processed.");
        }


        /// <summary>
        /// Process a batch of asynchronous orders from the queue and submit them to the database within a transaction
        /// </summary>
        private static void ProcessOrders()
        {
            OrderQueue orQueue = new OrderQueue();
            while (true)
            {
                int processedItems = 0;
                ArrayList queueOrders = new ArrayList();

                for (int j = 0; j < batchSize; j++)
                {
                    try
                    {
                        queueOrders.Add(orQueue.Receive());
                    }
                    catch (Exception)
                    {
                        j = batchSize;
                    }
                }

                //process the queued orders
                for (int k = 0; k < queueOrders.Count; k++)
                {
                    //orQueue.Insert((Order)queueOrders[k]);  PETSHOP是插入到数据库方法,以下改为写到日志文件
              Order o = (Order)queueOrders[k];
                    MyLog.WriteLog("订单号：" + o.OrderID + " 商品ID：" + o.GoodsID);
                    processedItems++;
                    totalOrdersProcessed++;
                }

                MessageBox.Show("(Thread Id " + Thread.CurrentThread.ManagedThreadId + ") batch finished, " + processedItems + " items");
            }
        }