微軟消息隊列-MicroSoft Message Queue(MSMQ)隊列的C#使用

目錄

• 定義的接口
• 接口實現
• 創建隊列工廠
• 寫入隊列
• 獲取消息

什麼是MSMQ

Message Queuing(MSMQ) 是微軟開發的消息中間件，可應用於程序內部或程序之間的異步通訊。主要的機制是：消息的發送者把本身想要發送的信息放入一個容器中（咱們稱之爲Message），而後把它保存至一個系統公用空間的消息隊列(Message Queue)中；本地或者是異地的消息接收程序再從該隊列中取出發給它的消息進行處理。下圖展現了這一流程

MSMQ隊列是一個可持久的隊列，所以沒必要擔憂不間斷地插入隊列會致使數據的丟失,在網站系統中不用擔憂網站掛掉後數據丟失問題。
MSMQ微軟消息隊列，這個是個很好的異步通訊技術。很是相似於咱們手機發短信，雖然對方關機了，但只要一開機通訊服務器仍然自動會傳送信息。因此MSMQ有這樣的時間監控技術.很實用.能夠確保通訊的穩定性.

使用MSMQ的基本流程

　　1. 啓動MSMQ服務，【控制面板】--【程序與功能】--【關閉/打開windows功能】--添加MSMQ功能，勾選所有選項。

控制面板->控制面板->全部控制面板項->程序和功能->選中安裝

　　2. 建立Message Queue隊列。
　　3. 打開Message Queue隊列。
　　4. 將消息發送至Message Queue隊列或者從Message Queue隊列中接收消息。
　　5. 關閉Message Queue隊列。

C# 中使用MSMQ

使用MessageQueue類操做MSMQ，其在System.Messaging命名空間下，須要添加引用

定義的接口

public interface IMessageSender<T> : IDisposable
{
    /// <summary>
    ///     發送消息
    /// </summary>
    /// <param name="message">消息對象</param>
    void SendMessage(T message);

    void SendMessages(List<T> message);

    /// <summary>
    ///     發送消息
    /// </summary>
    /// <param name="message">消息對象</param>
    /// <param name="label">消息標籤</param>
    void SendMessage(T message, string label);
}

接口實現

/// <summary>
///     消息隊列對象，由MQueueFactory建立指定路徑的隊列對象，可發送或批量接收消息。
/// </summary>
/// <typeparam name="T">消息隊列存儲的消息對象類型</typeparam>
public sealed class MQueue<T> : IDisposable, IMessageSender<T>, IMessageReceiver<T>
{
    public MQueue(MessageQueue mq, string user = "Everyone")
    {
        InnerQueue = mq;
        InnerQueue.Formatter = new XmlMessageFormatter(new[] { typeof(T) });

        InnerQueue.SetPermissions(user ?? "Everyone",
            MessageQueueAccessRights.GenericRead | MessageQueueAccessRights.DeleteMessage |
            MessageQueueAccessRights.DeleteQueue | MessageQueueAccessRights.DeleteJournalMessage);
    }
    #region IMessageSender
    /// <summary>
    ///     內部消息隊列對象
    /// </summary>
    private MessageQueue InnerQueue { get; set; }

    /// <summary>
    ///     發送消息
    /// </summary>
    /// <param name="message">消息對象</param>
    public void SendMessage(T message)
    {
        InnerQueue.Send(message);
    }

    public void SendMessages(List<T> message)
    {
        foreach (var item in message)
        {
            InnerQueue.Send(item);
        }
    }

    /// <summary>
    ///     發送消息
    /// </summary>
    /// <param name="message">消息對象</param>
    /// <param name="label">消息標籤</param>
    public void SendMessage(T message, string label)
    {
        try
        {
            InnerQueue.Send(message, label);
        }
        catch (Exception ex)
        {
            var path = InnerQueue.Path;
            InnerQueue = new MessageQueue(path);
        }
    }
    #endregion 

    #region IMessageReceiver
    /// <summary>
    /// 獲取隊列中全部消息
    /// </summary>
    /// <typeparam name="T">消息類型</typeparam>
    /// <param name="exTarget">異常時觸發</param>
    /// <returns></returns>
    public List<T> GetAllMessages<T>(MQExceptionTarget exTarget = null)
    {
        return GetMessagesByNum<T>(null, exTarget);
    }
    /// <summary>
    /// 獲取隊列中指定數量消息
    /// </summary>
    /// <typeparam name="T">消息類型</typeparam>
    /// <param name="num">一次最多取num條數據，默認取全部數據</param>
    /// <param name="exTarget">異常委託</param>
    /// <returns></returns>
    public List<T> GetMessagesByNum<T>(int? num = null, MQExceptionTarget exTarget = null)
    {
        var list = new List<T>();
        if (num.HasValue && num <= 0)
        {
            return list;
        }
        MessageEnumerator enumerator = InnerQueue.GetMessageEnumerator2();
        while (enumerator.MoveNext())
        {
            Message msg = enumerator.RemoveCurrent();
            enumerator.Reset();
            if (msg != null)
            {
                try
                {
                    list.Add((T)msg.Body);
                    if (num.HasValue && list.Count >= num)
                    {
                        break;
                    }
                }
                catch (Exception ex)
                {
                    if (exTarget != null)
                        exTarget(ex);
                }
            }
        }
        return list;
    }
    #endregion

    public void Dispose()
    {
        if (InnerQueue != null)
        {
            InnerQueue.Dispose();
        }
    }
}

創建隊列工廠

/// <summary>
///  消息隊列工廠，經過指定路徑建立或獲取相應隊列對象
/// </summary>
public class MQueueFactory
{
    /// <summary>
    ///     默認隊列路徑，在未指定路徑的狀況下，將建立並返回該路徑的消息隊列對象
    /// </summary>
    private const string DefaultPath = @".\private$\myQueue";

    /// <summary>
    ///     建立默認路徑的消息隊列對象
    /// </summary>
    /// <typeparam name="T">消息隊列存儲的消息對象類型</typeparam>
    /// <returns></returns>
    public static MQueue<T> Create<T>()
    {
        return Create<T>(DefaultPath);
    }

    /// <summary>
    ///     建立指定路徑的消息隊列路徑
    /// </summary>
    /// <typeparam name="T">消息隊列存儲的消息對象類型</typeparam>
    /// <param name="connStr">指定的消息隊列連接字符串 def:".\private$\myQueue"</param>
    /// <param name="autoCreate">不存在則建立隊列 遠程地址不能建立</param>
    /// <param name="user">得到隊列額外權限的我的、組或計算機</param>
    /// <param name="cacheKey">web中Cache鍵值</param>
    /// <returns></returns>
    public static MQueue<T> Create<T>(string connStr=@".\private$\myQueue", bool autoCreate = false, string user = "Everyone", string cacheKey = "MQCache")
    {
        string path = connStr ?? DefaultPath;
        HttpContext httpContext = HttpContext.Current;
        if (autoCreate && !MessageQueue.Exists(path))
        {
            MessageQueue.Create(path);
        }
        var mq = new MessageQueue(path);
        if (httpContext != null)
        {
            string key = "MQueue" + typeof(T).Name + cacheKey;
            if ((httpContext.Cache[key] == null))
            {
                httpContext.Cache[key] = new MQueue<T>(mq);
            }
            return httpContext.Cache[key] as MQueue<T>;
        }
        return new MQueue<T>(mq,user);
    }
}

寫入隊列

MQueueFactory.Create<string>(@".\private$\myQueue", autoCreate: true).SendMessage("我是寫入的數據~~~");

獲取消息

MQueueFactory.Create<string>(@".\private$\myQueue").GetAllMessages<string>();

代碼來源：https://www.cnblogs.com/morang/p/mqdemo-msmq.html