Asp.net 面向接口可扩展框架之消息队列组件

消息队列对大多数人应该比较陌生。但是要提到MQ听说过的人会多很多。MQ就是英文单词"Message queue"的缩写,翻译成中文就是消息队列(我英语差,翻译错了请告知)。

PS：话说国人熟悉MQ比消息队列多,是不是因为国人的外语水平高于国语水平好几个数量级

1、看一下度娘怎么解释消息队列

参考链接:消息队列_百度百科

度娘解释消息队列是在两台计算机间传输的,套句很时髦的说法就是用来做分布式传输的,是个很高大上的东西

2、我的看法稍有不同

我更追溯到“消息队列”的字面“本源”的意思。我认为消息队列就是消息的管理容器工具

消息队列可以在“两台计算机间传输”，也可以同一台计算机不同进行进程间传输,甚至是同一进程内“传输”

3、消息队列使用主要场景

我认为主要有两种,一种是排队先进先出(也有加优先级的),另一种是消息发布订阅模式,当然两种方式“复合”使用也是可以的

4、消息队列主要解决什么问题

我们写的程序偶尔出现一些“灵异”问题。除了一般的业务逻辑bug外,主要就是折腾服务器了。比如,web服务器cpu满载、数据库cpu满载、内存满载、磁盘IO满载、网络带宽满载等等

我总结为两点,计算密集型问题(cpu满载)和资源密集型问题(内存、磁盘、网络)

我们要优化程序需要知道到底是哪种问题,针对不同问题进行不同的优化，优化一般无非“开源”和“节流”两种手段。

　　“开源”：增加计算能力(含增加cpu和服务器)和增加资源

“节流”：减少“多余”的逻辑和资源消耗

现实中的情况很复杂,有的时候很简单的逻辑(但资源耗费严重)也能导致cpu满载,我们认为程序是在“等”资源,其实它在“等”资源的时候依然吞噬了大量的“cpu”,所以把计算和资源消耗“拆分”开很多情况下更加有效

增加了cpu如何用得上,如果我们的程序是单线程,就算增加到256个cpu对性能改善也用处不大,另外增加了服务器我们的逻辑如何还能保持完整,不同服务器的程序如何协同工作,那就是“消息队列”隆重登场的时候了。

前面太抽象,直接上例子了

一、消息订阅模式

1、消息订阅代码

        public static void Test()
        {
            Soldier[] soldiers = new Soldier[] { new Soldier { Name = "士兵1" }, new Soldier { Name = "士兵2" }, new Soldier { Name = "士兵3" } };
            Produce<int> producer = new Produce<int>() { Instance = new DirectiveAction() };

            SubscribeChannel<int> channel = new SubscribeChannel<int>() { MaxDegreeOfParallelism = 3 };
            channel.Init();
            channel.AddProducer(producer);
            foreach (var item in soldiers)
            {
                channel.AddConsumer(new SubscribeConsume<int>() { Instance = item });
            }
            while (producer.Run())
            {
            }
        }

解读一下:

A：定义了一个"队伍"(士兵(Soldier)数组,等待接受命令(订阅长官的命令))

B：定义了一个“长官”(命令的生产(发布)者)

C：定义了一个消息订阅频道(SubscribeChannel)并把生产者和消费者都添加进去

　　这里面要重点说一下频道的重要性,大家“通信”必须在相同频道才可以相互沟通。不在一个频道,人民解放军是不能听美国将军的命令的

另一个方面也说明可以通过增加不同频道来建立多个消息队列,频道独立于生产者和消费者,同一个生产者或者消费者也可能同时为多个消息队列服务,Very Good！

D：不停的等待“长官”发布命令直至“长官”自己终止

2、F5

Asp.net 面向接口可扩展框架之消息队列组件

3、“指令”什么鬼,再看一些代码

    /// <summary>
    /// 命令操作
    /// </summary>
    public class DirectiveAction : IResultInstance<int>
    {
        /// <summary>
        /// 
        /// </summary>
        /// <param name="result"></param>
        /// <returns></returns>
        public bool Run(ref int result)
        {
            System.Threading.Thread.Sleep(2000);
            Console.Write(string.Concat(string.Concat(DateTime.Now.ToString("hh:mm:ss.fff"), " 请输入指令:")));
            string str = Console.ReadLine();
            Console.WriteLine(string.Concat("I say ", str));
            if (string.Equals(str, "Exit", StringComparison.CurrentCultureIgnoreCase))
                return false;
            result = IntConverter.Instance.Get(str);
            return true;
        }
        public void OnSuccess()
        {
        }
        public void OnFail()
        {
        }
        public void OnException(Exception ex)
        {
        }
    }

命令操作