关于我们

质量为本、客户为根、勇于拼搏、务实创新

< 返回新闻公共列表

C#队列学习笔记:MSMQ入门二

发布时间:2020-03-28 00:00:00

       一、引言

    按照专用队列解释: MachineName\Private$\QueueName,只针对于本机的程序才可以调用的队列,有些情况下为了安全起见定义为私有队列。所以刚开始的时候认为,要想访问远程消息队列,只能使用公共队列。但是后来发现,公共队列依赖Domain Controller(域控),在实际部署的时候,要求使用消息队列的应用一定要在某个域中,有些太苛刻!后来发现,私有队列也是可以远程访问的。(很困惑为什么私有队列只能本地访问,这句话,到处都能看到?!)

    二、工作组下的本地C/S     2.1、项目建立

    新建4个项目:

    2.2、项目代码

    2.2.1、Model项目

    /// /// 消息队列实体///     [Serializable]public class MqMessage
    {/// /// 对应Message的Label/// public string Label { get; set; }/// /// 对应Message的Body,CommandType为操作类型,List为操作列表。/// public Dictionary<CommandType, List<string>> Body { get; set; } = new Dictionary<CommandType, List<string>>();/// /// 无参构造函数/// public MqMessage()
        {
        }/// /// 有参构造函数/// /// /// public MqMessage(string label, Dictionary<CommandType, List<string>> body)
        {
            Label = label;
            Body = body;
        }
    }/// /// 操作类型/// public enum CommandType
    {
        Create = 1, //创建Update = 2, //更新Delete = 3  //删除}
MqMessage.cs

    2.2.2、Common项目

    /// /// 日志帮助类/// public static class LogHelper
    {private static readonly string errLogSavePath = ConfigurationManager.AppSettings["ErrLogSavePath"] ?? AppDomain.CurrentDomain.BaseDirectory;/// /// 异常日志方法重载/// /// 异常信息public static void WriteLog(Exception ex)
        {
            WriteLog(GetErrMsg(ex));
        }/// /// 异常日志方法重载/// /// 日志内容public static void WriteLog(string message)
        {
            WriteLog(errLogSavePath, message);
        }/// /// 异常日志方法重载/// /// 日志文件路径/// 日志内容public static void WriteLog(string filepath, string message)
        {try{if (!Directory.Exists(filepath))
                {
                    Directory.CreateDirectory(filepath);
                }string filename = DateTime.Now.ToString("yyyy-MM-dd") + ".txt";using (StreamWriter sw = new StreamWriter(filepath + "\\" + filename, true))
                {
                    sw.WriteLine("--------------------------------------------");
                    sw.WriteLine($"{DateTime.Now.ToLongTimeString()}:{DateTime.Now.Millisecond}\t{message}");
                    sw.Close();
                }
            }catch (Exception ex)
            {throw new Exception(GetErrMsg(ex));
            }
        }/// /// 获取异常详细信息/// /// /// private static string GetErrMsg(Exception ex)
        {string errMessage = "";for (Exception tempException = ex; tempException != null; tempException = tempException.InnerException)
            {
                errMessage += tempException.Message + Environment.NewLine + Environment.NewLine;
            }
            errMessage += ex.ToString();return errMessage;
        }
    }
LogHelper.cs
    /// /// 消息队列管理器/// public class MqManager : IDisposable
    {private MessageQueue _mq = null;private readonly LinkType linkType = LinkType.LocalHost;    //链接类型,远程时使用LinkType.RemoteServer。private readonly string remoteServer = "192.168.2.165";     //远程服务器IP地址public static MqManager LinkServer { get; } = new MqManager();/// /// 初始化函数/// /// 链接类型public void MqManagerInit(LinkType linkType)
        {if (_mq == null)
            {string _path;if (linkType == LinkType.LocalHost)
                {
                    _path = @".\private$\" + (ConfigurationManager.AppSettings["MSMQName"] ?? "HelloWorld");
                }else{
                    _path = "FormatName:DIRECT=TCP:" + remoteServer + @"\private$\" + (ConfigurationManager.AppSettings["MSMQName"] ?? "HelloWorld");
                }
                _mq = new MessageQueue(_path)
                {
                    Formatter = new BinaryMessageFormatter()
                };
            }
        }/// /// 有参构造函数/// public MqManager()
        {
            MqManagerInit(linkType);
        }/// /// 发送消息队列(事务)/// /// public void Send(MqMessage message)
        {
            MessageQueueTransaction transaction = new MessageQueueTransaction();
            transaction.Begin();
            _mq.Send(message.Body, message.Label, transaction);
            transaction.Commit();
        }/// /// 接收消息队列/// /// public Message Receive()
        {
            Message msg = null;try{
                msg = _mq.Receive(new TimeSpan(0, 0, 1));
            }catch (Exception ex)
            {throw new Exception(ex.Message);
            }return msg;
        }/// /// 释放资源/// public void Dispose()
        {if (_mq != null)
            {
                _mq.Close();
                _mq.Dispose();
                _mq = null;
            }
        }
    }/// /// 链接类型/// public enum LinkType
    {
        LocalHost = 1,      //本地服务器RemoteServer = 2    //远程服务器}
MqManager.cs

    2.2.3、Send项目

    class Program
    {static void Main(string[] args)
        {
            MqMessage mqMessage = new MqMessage();
            List<string> list = new List<string>();

            Console.WriteLine("请输入内容按回车发送,多个内容请用英文逗号隔开,退出请输入Exit。");string receiveKey = Console.ReadLine();while (receiveKey.ToLower() != "exit")
            {if (receiveKey.Length > 0)
                {
                    mqMessage.Label = Guid.NewGuid().ToString();

                    list.Clear();
                    list = receiveKey.Split(new char[] { ',' }).ToList();
                    mqMessage.Body.Clear();
                    mqMessage.Body.Add(CommandType.Create, list);try{
                        MqManager.LinkServer.Send(mqMessage);
                        Console.WriteLine("内容已发送成功。");
                    }catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                        LogHelper.WriteLog(ex);
                    }
                }
                receiveKey = Console.ReadLine();
            }

            MqManager.LinkServer.Dispose();
        }
    }
Program.cs

    2.2.4、Receive项目

    /// /// 接收消息队列管理(线程)/// public class ReceiveManager : IDisposable
    {private Thread _thread = null;public static ReceiveManager Instance { get; set; } = new ReceiveManager();/// /// 开始/// public void Start()
        {
            StartReceive();
        }/// /// 接收线程/// private void StartReceive()
        {
            _thread = new Thread(new ThreadStart(Receive))
            {
                Name = "ReceiveThread",
                IsBackground = true};
            _thread.Start();
        }/// /// 接收线程调用方法/// private void Receive()
        {
            Message msg = null;while (true)
            {try{
                    msg = MqManager.LinkServer.Receive();if (msg != null)
                    {
                        Console.WriteLine("----------------------------------------------------");
                        Console.WriteLine("Lable: " + msg.Label);
                        Dictionary<CommandType, List<string>> keyValuePairs = msg.Body as Dictionary<CommandType, List<string>>;
                        Console.WriteLine("Body CommandType: " + keyValuePairs.Keys.First());
                        Console.WriteLine("Body Details: ");foreach (var item in keyValuePairs.Values.First())
                        {
                            Console.WriteLine(item);
                        }
                        Console.WriteLine("----------------------------------------------------");
                    }
                }catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                    LogHelper.WriteLog(ex);
                }
                Thread.Sleep(1000);
            }
        }/// /// 结束/// public void Stop()
        {
            Dispose();
        }/// /// 释放资源/// public void Dispose()
        {try{if (_thread != null)
                {
                    _thread.Abort();
                    _thread.Join();
                    _thread = null;
                }

                MqManager.LinkServer.Dispose();
            }catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }
    }
ReceiveManage.cs
    class Program
    {static void Main(string[] args)
        {
            ReceiveManager.Instance.Start();
            Console.WriteLine("退出请输入Exit");string receiveKey = Console.ReadLine();while (receiveKey.ToLower() != "exit")
            {
                receiveKey = Console.ReadLine();
            }
            ReceiveManager.Instance.Stop();
            Console.Read();
        }
    }
Program.cs

    2.3、运行测试

    客户端发送hello,world:

    服务端接收到的信息:

    三、工作组下的远程C/S     3.1、代码调整

    工作组下的远程C/S,代码已经在上面的示例中提供,将Common\MqManager.cs下的:

    private readonly LinkType linkType = LinkType.LocalHost;改成private readonly LinkType linkType = LinkType.RemoteServer;即可。

    3.2、访问权限

    既然要与远程服务器交互(发送/接收)队列信息,首当其冲的是访问权限问题,没有权限,一切免谈。

    下面讲一下远程服务器(代码中的192.168.2.165,Win7系统)要设置的内容:

    3.2.1、在运行中输入compmgmt.msc->服务和应用程序->消息队列->右键属性->服务器安全性->禁用未经身份验证的 RPC 调用->把勾勾去掉->应用。

    3.2.2、在消息队列->专用队列->新建一个代码中用到的HelloWorld队列,勾上事务性->确定。

    为什么要手工建HelloWorld消息队列?因为要对这个队列进行匿名访问授权,后面会讲到。至于事务性这个勾,这个要与代码相一致。因为本示例中使用了MessageQueueTransaction来发送事务信息,所以必须得勾上这个勾,不然的话,发送时没有任何的报错信息,但是服务器就是收不到队列信息。

    3.2.3、专用队列->HelloWorld->右键属性->安全->ANONYMOUS LOGON->完全控制->应用。

    3.2.4、在运行中输入regedit->HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\MSMQ\Parameters\security->新建两个DWORD值:AllowNonauthenticatedRpc、NewRemoteReadServerDenyWorkgroupClient->分别双击将数值数据改成1。

    3.2.5、关于防火墙,我是关闭了的,假如您的电脑防火墙是打开了的话,请检查一下Message Queuing是不是被允许的?

    3.3、运行测试

    客户端发送A,B,C,D:

    服务器端接收到的信息:

 

    参考自:

    https://www.cnblogs.com/xinhaijulan/archive/2010/08/22/1805768.html

    https://www.cnblogs.com/minily/p/7397746.html

    https://blog.csdn.net/jiyiqinlovexx/article/details/17803857

    https://www.cnblogs.com/mmbbflyer/p/7773303.html


/template/Home/Zkeys/PC/Static