【c#】队列(Queue)和MSMQ(消息队列)的基础使用

简介: 原文:【c#】队列(Queue)和MSMQ(消息队列)的基础使用      首先我们知道队列是先进先出的机制,所以在处理并发是个不错的选择。然后就写两个队列的简单应用。 Queue 命名空间     命名空间:System.Collections,不在这里做过多的理论解释,这个东西非常的好理解。
原文: 【c#】队列(Queue)和MSMQ(消息队列)的基础使用

 

    首先我们知道队列是先进先出的机制,所以在处理并发是个不错的选择。然后就写两个队列的简单应用。

Queue

命名空间

    命名空间:System.Collections,不在这里做过多的理论解释,这个东西非常的好理解。

    可以看下官方文档:https://docs.microsoft.com/zh-cn/dotnet/api/system.collections.queue?view=netframework-4.7.2

示例代码

我这里就是为了方便记忆做了一个基本的例子,首先创建了QueueTest类:

包含了获取队列的数量,入队和出队的实现

 1  public class QueueTest
 2     {
 3         public static Queue<string> q = new Queue<string>();
 4 
 5         #region 获取队列数量
 6         public int GetCount()
 7         {
 8 
 9             return q.Count;
10         }
11         #endregion
12 
13         #region 队列添加数据
14         public void IntoData(string qStr)
15         {
16             string threadId = System.Threading.Thread.CurrentThread.ManagedThreadId.ToString();
17             q.Enqueue(qStr);
18             Console.WriteLine($"队列添加数据: {qStr};当前线程id:{threadId}");
19         }
20         #endregion
21 
22         #region 队列输出数据
23 
24         public string OutData()
25         {
26             string threadId = System.Threading.Thread.CurrentThread.ManagedThreadId.ToString();
27             string str = q.Dequeue();
28             Console.WriteLine($"队列输出数据: {str};当前线程id:{threadId}");
29             return str;
30         }
31         #endregion
32 
33     }

为了模拟并发情况下也不会出现重复读取和插入混乱的问题所以写了TaskTest类里面开辟了两个异步线程进行插入和读取:

 1     class TaskTest
 2     {
 3 
 4         #region 队列的操作模拟
 5         public static void QueueMian()
 6         {
 7             QueueA();
 8             QueueB();
 9         }
10         private static async void QueueA()
11         {
12             QueueTest queue = new QueueTest();
13             var task = Task.Run(() =>
14             {
15                 for (int i = 0; i < 20; i++)
16                 {
17                     queue.IntoData("QueueA" + i);
18                 }
19             });
20             await task;
21             Console.WriteLine("QueueAA插入完成,进行输出:");
22 
23             while (queue.GetCount() > 0)
24             {
25                 queue.OutData();
26             }
27         }
28 
29         private static async void QueueB()
30         {
31             QueueTest queue = new QueueTest();
32             var task = Task.Run(() =>
33             {
34                 for (int i = 0; i < 20; i++)
35                 {
36                     queue.IntoData("QueueB" + i);
37                 }
38             });
39             await task;
40             Console.WriteLine("QueueB插入完成,进行输出:");
41 
42             while (queue.GetCount() > 0)
43             {
44                 queue.OutData();
45             }
46         }
47         #endregion
48 
49     }

效果展示

然后在main函数直接调用即可:

通过上面的截图可以看出插入线程是无先后的。

这张去除也是线程无先后,但是数据是根据插入的数据顺序取的,也就是说多线程取随便取,但是取的数据是根据插入的顺序取值。

MSMQ

msmq是微软提供的消息队列,本来在windows系统中就存在,但是默认没有开启。需要开启。

开启安装

打开控制面板=>程序和功能=> 启动或关闭windows功能 => Microsoft Message Queue(MSMQ)服务器=>Microsoft Message Queue(MSMQ)服务器核心

一般选择:MSMQ Active Directory域服务继承和MSMQ HTTP支持即可。

点击确定等待安装成功。

命名空间

需要引用System.Messaging.DLL

命名空间:System.Messaging

官方资料文档:https://docs.microsoft.com/zh-cn/dotnet/api/system.messaging.messagequeue?view=netframework-4.7.2

示例代码

与上面queue同样的示例方式,创建一个MSMQ类,实现创建消息队列,查询数据,入列,出列功能:

  1  /// <summary>
  2     /// MSMQ消息队列
  3     /// </summary>
  4     class MSMQ
  5     {
  6         static string path = ".\\Private$\\myQueue";
  7         static MessageQueue queue;
  8         public static void Createqueue(string queuePath)
  9         {
 10             try
 11             {
 12                 if (MessageQueue.Exists(queuePath))
 13                 {
 14                     Console.WriteLine("消息队列已经存在");
 15                     //获取这个消息队列
 16                     queue = new MessageQueue(queuePath);
 17                 }
 18                 else
 19                 {
 20                     //不存在,就创建一个新的,并获取这个消息队列对象
 21                     queue = MessageQueue.Create(queuePath);
 22                     path = queuePath;
 23                 }
 24             }
 25             catch (Exception e)
 26             {
 27                 Console.WriteLine(e.Message);
 28             }
 29 
 30         }
 31 
 32 
 33         #region 获取消息队列的数量
 34         public static int GetMessageCount()
 35         {
 36             try
 37             {
 38                 if (queue != null)
 39                 {
 40                     int count = queue.GetAllMessages().Length;
 41                     Console.WriteLine($"消息队列数量:{count}");
 42                     return count;
 43                 }
 44                 else
 45                 {
 46                     return 0;
 47                 }
 48             }
 49             catch (MessageQueueException e)
 50             {
 51 
 52                 Console.WriteLine(e.Message);
 53                 return 0;
 54             }
 55 
 56 
 57         }
 58         #endregion
 59 
 60         #region 发送消息到队列
 61         public static void SendMessage(string qStr)
 62         {
 63             try
 64             {
 65                 //连接到本地队列
 66 
 67                 MessageQueue myQueue = new MessageQueue(path);
 68 
 69                 //MessageQueue myQueue = new MessageQueue("FormatName:Direct=TCP:192.168.12.79//Private$//myQueue1");
 70 
 71                 //MessageQueue rmQ = new MessageQueue("FormatName:Direct=TCP:121.0.0.1//private$//queue");--远程格式
 72 
 73                 Message myMessage = new Message();
 74 
 75                 myMessage.Body = qStr;
 76 
 77                 myMessage.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
 78 
 79                 //发生消息到队列中
 80 
 81                 myQueue.Send(myMessage);
 82 
 83                 string threadId = System.Threading.Thread.CurrentThread.ManagedThreadId.ToString();
 84                 Console.WriteLine($"消息发送成功: {qStr};当前线程id:{threadId}");
 85             }
 86             catch (MessageQueueException e)
 87             {
 88                 Console.WriteLine(e.Message);
 89             }
 90         }
 91         #endregion
 92 
 93         #region 连接消息队列读取消息
 94         public static void ReceiveMessage()
 95         {
 96             MessageQueue myQueue = new MessageQueue(path);
 97 
 98 
 99             myQueue.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
100 
101             try
102 
103             {
104 
105                 //从队列中接收消息
106 
107                 Message myMessage = myQueue.Receive(new TimeSpan(10));// myQueue.Peek();--接收后不消息从队列中移除
108                 myQueue.Close();
109 
110                 string context = myMessage.Body.ToString();
111                 string threadId = System.Threading.Thread.CurrentThread.ManagedThreadId.ToString();
112                 Console.WriteLine($"--------------------------消息内容: {context};当前线程id:{threadId}");
113 
114             }
115 
116             catch (System.Messaging.MessageQueueException e)
117 
118             {
119 
120                 Console.WriteLine(e.Message);
121 
122             }
123 
124             catch (InvalidCastException e)
125 
126             {
127 
128                 Console.WriteLine(e.Message);
129 
130             }
131 
132         }
133         #endregion
134     }

这里说明一下path这个字段,这是消息队列的文件位置和队列名称,我这里写的“.”(点)就是代表的位置MachineName字段,,代表本机的意思

然后TaskTest类修改成这个样子:

 1 class TaskTest
 2     {
 3 
 4         #region 消息队列的操作模拟
 5         public static void MSMQMian()
 6         {
 7             MSMQ.Createqueue(".\\Private$\\myQueue");
 8             MSMQA();
 9             MSMQB();
10             Console.WriteLine("MSMQ结束");
11         }
12         private static async void MSMQA()
13         {
14             var task = Task.Run(() =>
15             {
16                 for (int i = 0; i < 20; i++)
17                 {
18                     MSMQ.SendMessage("MSMQA" + i);
19                 }
20             });
21             await task;
22             Console.WriteLine("MSMQA发送完成,进行读取:");
23 
24             while (MSMQ.GetMessageCount() > 0)
25             {
26                 MSMQ.ReceiveMessage();
27             }
28         }
29 
30         private static async void MSMQB()
31         {
32             var task = Task.Run(() =>
33             {
34                 for (int i = 0; i < 20; i++)
35                 {
36                     MSMQ.SendMessage("MSMQB" + i);
37                 }
38             });
39             await task;
40             Console.WriteLine("MSMQB发送完成,进行读取:");
41 
42             while (MSMQ.GetMessageCount() > 0)
43             {
44                 MSMQ.ReceiveMessage();
45             }
46         }
47         #endregion

 效果展示

本机查看消息队列

创建成功的消息队列我们可以在电脑上查看:我的电脑=>管理 =>计算机管理 =>服务与应用程序 =>消息队列 =>专用队列就看到我刚才创建的消息队列

 

目录
相关文章
|
消息中间件 存储 负载均衡
拆解一下消息队列、任务队列、任务调度系统
拆解一下消息队列、任务队列、任务调度系统
488 0
|
6月前
|
消息中间件 NoSQL Java
别再用 Redis List 实现消息队列了,Stream 专为队列而生
别再用 Redis List 实现消息队列了,Stream 专为队列而生
139 0
|
3月前
|
消息中间件 存储 负载均衡
现代消息队列与云存储问题之消息队列支持定时消息和延迟队列的问题如何解决
现代消息队列与云存储问题之消息队列支持定时消息和延迟队列的问题如何解决
|
27天前
|
消息中间件 C# 数据安全/隐私保护
手写MSMQ微软消息队列收发工具类
【10月更文挑战第16天】本示例展示了如何使用C#语言编写MSMQ(微软消息队列)的收发工具类,包括发送和接收消息的方法。通过检查队列是否存在并创建、使用`MessageQueue`类发送和接收消息。示例还提供了简单的调用方式,并提醒用户注意权限管理和异常处理。
|
3月前
|
消息中间件 存储 Kafka
现代消息队列与云存储问题之Kafka在海量队列场景下存在性能的问题如何解决
现代消息队列与云存储问题之Kafka在海量队列场景下存在性能的问题如何解决
|
5月前
|
消息中间件
【消息队列开发】 虚拟主机设计——放送消息到队列/交换机中
【消息队列开发】 虚拟主机设计——放送消息到队列/交换机中
|
5月前
|
消息中间件 安全
【消息队列开发】 虚拟主机设计——操作队列
【消息队列开发】 虚拟主机设计——操作队列
|
5月前
|
消息中间件 存储 Java
消息队列-死信队列
消息队列-死信队列
52 0
|
5月前
|
消息中间件 存储 负载均衡
消息队列 MQ产品使用合集之如何排查是哪个队列导致的异常TPS增加
阿里云消息队列MQ(Message Queue)是一种高可用、高性能的消息中间件服务,它允许您在分布式应用的不同组件之间异步传递消息,从而实现系统解耦、流量削峰填谷以及提高系统的可扩展性和灵活性。以下是使用阿里云消息队列MQ产品的关键点和最佳实践合集。
|
消息中间件 存储 监控
消息队列与任务队列的区别
消息队列和任务队列是我们在软件系统中常常遇到的概念。尽管它们的名字相似,但实际上它们有不同的用途和工作原理。本文将介绍消息队列和任务队列之间的区别。
567 0

热门文章

最新文章