消息队列MSMQ

消息队列

消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式，软件的贮列用来处理一系列的输入，通常是来自使用者。消息队列提供了异步的通信协议，每一个贮列中的纪录包含详细说明的资料，包含发生的时间，输入装置的种类，以及特定的输入参数，也就是说：消息的发送者和接收者不需要同时与消息队列互交。消息会保存在队列中，直到接收者取回它。

简单的说队列就是贮存了我们需要处理的Command但是并不是及时的拿到其处理结果；

win7中安装消息队列组件，其他操作系统请百度搜索相关资料。



如果服务没有自动启动，需要启动服务：



示例：

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Messaging;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleApplication39
{
class Program
{
static void Main(string[] args)
{
Dequeue();
Enqueue();
Console.Read();
}
static void Enqueue()
{
SystemMessageQueue<Person> queue = SystemMessageQueue<Person>.Install("PersonQueue");
Task.Run(() =>
{
var i = 10;
while (true)
{
queue.Enqueue(new Person() { age = i, name = "这是入队消息：name" + i });
i++;
Thread.Sleep(1000);
}
});
}
static void Dequeue()
{
SystemMessageQueue<Person> queue = SystemMessageQueue<Person>.Install("PersonQueue");
Task.Run(() =>
{
while (true)
{
string result = queue.Dequeue()?.ToString();
if (!string.IsNullOrEmpty(result)) Console.WriteLine(result);
Thread.Sleep(500);
}
});
}
}
public class Person
{
public int age { get; set; }
public string name { get; set; }
public override string ToString() => string.Format("age:{0} name:{1}", age, name);
}
/// <summary>
/// 系统消息队列
/// </summary>
/// <typeparam name="T">消息类型</typeparam>
public class SystemMessageQueue<T>
{
private string _queuename = string.Empty;
private static SystemMessageQueue<T> _queue = null;
private MessageQueue _msmq = null;
private SystemMessageQueue()
{
}
/// <summary>
/// 初始化队列
/// </summary>
/// <param name="queuename">队列名称</param>
/// <returns>队列实例</returns>
public static SystemMessageQueue<T> Install(string queuename)=> _queue = new SystemMessageQueue<T>(queuename);
private SystemMessageQueue(string queuename)
{
this._queuename = queuename;
string path = ".\\private$\\" + queuename;
if (!MessageQueue.Exists(path))
MessageQueue.Create(path);
_msmq = new MessageQueue(path);
_msmq.Formatter = new System.Messaging.XmlMessageFormatter(new Type[] { typeof(T) });
}

/// <summary>
/// 队列名称
/// </summary>
public string QueueName => _queuename;
/// <summary>
/// 入队
/// </summary>
/// <param name="msg">入队消息</param>
public void Enqueue(T msg) => _msmq.Send(msg);
/// <summary>
/// 出队(从队列的开头移除的对象，如果队列中没有消息则返回类型T的默认值)
/// </summary>
/// <returns>从队列的开头移除的对象，如果队列中没有消息则返回类型T的默认值。</returns>
public T Dequeue() => GetMsgCount() > 0 ? (T)_msmq.Receive(new TimeSpan(5)).Body :default(T);
/// <summary>
/// 出队(返回队列开头处的对象但不移除,如果队列中没有消息则返回类型T的默认值)
/// </summary>
/// <returns>返回队列开头处的对象但
c2dd
不移除,如果队列中没有消息则返回类型T的默认值</returns>
public T Peek() => GetMsgCount() > 0 ? (T)_msmq.Peek().Body : default(T);

/// <summary>
/// 出队(从队列的开头移除的对象，若无数据线程阻塞，直至有数据)
/// </summary>
/// <returns>从队列的开头移除的对象，若无数据线程阻塞，直至有数据。</returns>
public T BlockingDequeue() => (T)_msmq.Receive().Body;

/// <summary>
/// 获取队列中消息数量
/// </summary>
/// <returns>队列中的消息数量</returns>
public int GetMsgCount() => _msmq.GetAllMessages().Length;
}
}