針對高並發，可擴展的互聯網架構，搭建消息隊列（一）

針對高並發，可擴展的互聯網架構，搭建消息隊列（一）

　　想開發高並發可擴展的互聯網架構，消息隊列是不可缺少的，目前主流的消息隊列，有windows自帶的MSMQ，還有跨平台的強大的ZeroMQ,這里我們就選用ZeroMQ.

　　ZeroMQ介紹：（也拼寫作 ØMQ、 0MQ 或 ZMQ） 是個非常輕量級的開源消息隊列軟件。它沒有獨立的服務器，消息直接從一個應用程序被發送到另一個應用程序。ZeroMQ的學習和應用也非常簡單，它只有一個 C++ 編寫成的單個庫文件libzmq.dll， 可以鏈接到任何應用程序中。如果要在.NET 環境中使用，我們需要用到一個C#編寫的名為 clrzmq.dll 包裝庫。ZeroMQ可以在 Windows、 OS X 和 Linux 等多種操作系統上運行， C、 C++、C#、 Java、 Python 等語言都可以編寫ZeroMQ 應用程序這使得不同平台上的不同應用程序之間可以相互通訊。

1、環境搭建：

　　codeproject專題，下載對應的Download binaries - 377.6 KB，解壓縮到你的指定路徑。

　　這里我們就不詳細介紹，主要說一下C#封裝好的版本，NetMQ，是基於ZeroMQ進行封裝的。就不需要下載了，直接nuget上獲取：

　　PM> Install-Package NetMQ

　　為什么不直接用ZeroMQ，而使用NetMQ，運行非托管代碼的托管應用程序內可能會出現許多想不到的問題，像內存泄漏和奇怪的沒有訪問錯誤。而NetMQ使用原生的C#語言，它更容易調試原生C#代碼，你可以下載代碼，調試你的系統。你可以在github上貢獻。

　　待安裝好后，系統會自動添加NetMQ的引用。

　　

　　可以看到，NetMQ是基於zmq進行開發的，其實就是ZeroMQ了，並且已經為我們封裝了各種功能的MQ對象，比如REP/REQ ,PUB/SUB(主題式訂閱),XPUB/XSUB（非主題訂閱），Push/Pull，甚至還有路由模式等，從字面意義上，應該能看出個大概，后面我們一個一個進行測試使用。

　　先看個簡單的demo，初步了解一下：

　　

class Program
{
  static void Main(string[] args)
  {
    using (NetMQContext context = NetMQContext.Create())
    {
      Task serverTask = Task.Factory.StartNew(() =>Server(context));
      Task clientTask = Task.Factory.StartNew(() => Client(context));
      Task.WaitAll(serverTask, clientTask);
    }
  }
 
  static void Server(NetMQContext context)
  {
    using (NetMQSocket serverSocket = context.CreateResponseSocket())
    {
      serverSocket.Bind("tcp://*:5555");
 
      while (true)
      {
        string message = serverSocket.ReceiveString();
 
        Console.WriteLine("Receive message {0}", message);
 
        serverSocket.Send("World");          
 
        if (message == "exit")
        {
          break;
        }
      }
    }      
  }
 
  static void Client(NetMQContext context)
  {
    using (NetMQSocket clientSocket = context.CreateRequestSocket())
    {
      clientSocket.Connect("tcp://127.0.0.1:5555");
 
      while (true)
      {
        Console.WriteLine("Please enter your message:");
        string message = Console.ReadLine();
        clientSocket.Send(message);
 
        string answer = clientSocket.ReceiveString();
 
        Console.WriteLine("Answer from server: {0}", answer);
 
        if (message == "exit")
        {
          break;
        }
      }
    }
  }
}

　　代碼比較簡潔的介紹了REP/REQ模式下NetMQ的使用，而且我們可以看到，這個Mq對象是可以在不同的線程間切換使用的，也許你會測試中文，那就先序列化再反序列化吧，因為可能會出現亂碼喲。

　　這里，我先簡單根據NetMQ,封裝一個Server端和一個Client端，方便后面使用，當然也可以不封裝，直接使用：

　　Server：

　　

/// <summary>
    /// Mq服務端
    /// </summary>
    public class OctMQServer : IDisposable
    {
        public event EventHandler<DataEventArgs<NetMQSocket, NetMQMessage>> OnReceive;

        protected virtual void OnOnReceive(DataEventArgs<NetMQSocket, NetMQMessage> e)
        {
            EventHandler<DataEventArgs<NetMQSocket, NetMQMessage>> handler = OnReceive;
            if (handler != null) handler(this, e);
        }

        private int _port;
        private NetMQSocket _serverSocket;
        private ServerType _type;
        private NetMQContext _context;

        public void Init(int port, ServerType type)
        {
            _type = type;
            _port = port;
            _context = NetMQContext.Create();
            CreateServer();
        }

        void CreateServer()
        {
            switch (_type)
            {
                case ServerType.Response:
                    _serverSocket = _context.CreateResponseSocket(); break;
                case ServerType.Pub:
                    _serverSocket = _context.CreatePushSocket(); break;
                case ServerType.Router:
                    _serverSocket = _context.CreateRouterSocket(); break;
                case ServerType.Stream:
                    _serverSocket = _context.CreateStreamSocket(); break;
                case ServerType.Push:
                    _serverSocket = _context.CreatePushSocket(); break;
                case ServerType.XPub:
                    _serverSocket = _context.CreateXPublisherSocket(); break;
                default:
                    _serverSocket = _context.CreateResponseSocket(); break;
            }
            _serverSocket.Bind("tcp://*:" + _port);
            Task.Factory.StartNew(() =>
            AsyncRead(_serverSocket), TaskCreationOptions.LongRunning);
        }

        private void AsyncRead(NetMQSocket serverSocket)
        {
            while (true)
            {
                var msg = serverSocket.ReceiveMessage();
                OnOnReceive(new DataEventArgs<NetMQSocket, NetMQMessage>(serverSocket, msg));
            }
        }


        public NetMQSocket Server
        {
            get { return _serverSocket; }
        }

        public void Dispose()
        {
            _serverSocket.Dispose();
            _context.Dispose();
        }

        public void Send(NetMQMessage msg)
        {
            _serverSocket.SendMessage(msg);
        }

        public NetMQMessage CreateMessage()
        {
            return new NetMQMessage();
        }
    }

　　這樣，使用者就可以根據枚舉進行服務端的創建， 不用糾結到底用哪一種服務端，並且封裝了一些消息的異步事件，方便在開發中使用，可以使用多播委托，針對不同的消息進行不同的處理，我這里使用的while循環，當然，在netmq內部提供了循環器和心跳等，都可以在實際的開發中進行擴展和使用：Poller和NetMQTimer。

 

　　Client：

　　

 /// <summary>
    /// MQ客戶端
    /// </summary>
    public class OctMQClient:IDisposable
    {
        public event EventHandler<DataEventArgs<NetMQSocket, NetMQMessage>> OnReceive;

        protected virtual void OnOnReceive(DataEventArgs<NetMQSocket, NetMQMessage> e)
        {
            EventHandler<DataEventArgs<NetMQSocket, NetMQMessage>> handler = OnReceive;
            if (handler != null) handler(this, e);
        }

        private int _port;
        private NetMQSocket _clientSocket;
        private ClientType _type;
        private NetMQContext _context;
        private string _ip;
        private Task task;
        public void Init(string ip, int port, ClientType type)
        {
            _type = type;
            _ip = ip;
            _port = port;
            _context = NetMQContext.Create();
            CreateClient();
        }

        void CreateClient()
        {
            switch (_type)
            {
                case ClientType.Request:
                    _clientSocket = _context.CreateRequestSocket(); break;
                case ClientType.Sub:
                    _clientSocket = _context.CreateSubscriberSocket(); break;
                case ClientType.Dealer:
                    _clientSocket = _context.CreateDealerSocket(); break;
                case ClientType.Stream:
                    _clientSocket = _context.CreateStreamSocket(); break;
                case ClientType.Pull:
                    _clientSocket = _context.CreatePullSocket(); break;
                case ClientType.XSub:
                    _clientSocket = _context.CreateXSubscriberSocket(); break;
                default:
                    _clientSocket = _context.CreateRequestSocket(); break;
            }
            _clientSocket.Connect("tcp://" + _ip + ":" + _port);
        }

        public void StartAsyncReceive()
        {
            task = Task.Factory.StartNew(() =>
         AsyncRead(_clientSocket), TaskCreationOptions.LongRunning);
           
        }

        private void AsyncRead(NetMQSocket cSocket)
        {
            while (true)
            {
                var msg = cSocket.ReceiveMessage();
                OnOnReceive(new DataEventArgs<NetMQSocket, NetMQMessage>(cSocket, msg));
            }
        }

        public NetMQSocket Client
        {
            get { return _clientSocket; }
        }

        public T GetClient<T>() where T : NetMQSocket
        {
            return (T)_clientSocket;
        }

        public void Send(NetMQMessage msg)
        {
            _clientSocket.SendMessage(msg);
        }

        public NetMQMessage CreateMessage()
        {
            return new NetMQMessage();
        }

        public NetMQMessage ReceiveMessage()
        {
            return _clientSocket.ReceiveMessage();
        }

        public void Dispose()
        {
            _clientSocket.Dispose();
            _context.Dispose();
            if (task != null)
            {
                task.Dispose();
            }
        }
    }

 

　　客戶端提供了，同步接受消息和異步接收消息兩種方式，當啟動異步時，就開始循環的讀取消息了，當讀到消息時拋出事件，並且針對任務等做了資源的釋放。並提供創建消息和返回MQ對象等公共方法，可以在開發過程中快速的入手和使用。

　　先簡單說一下response和request模式，就是響應模式，當mq客戶端向mq的服務端發送消息時，需要得到及時的響應，並返回給使用者或者是用戶，這就需要及時響應的服務端程序，一般的MQ都會有這種功能，也是使用最廣泛的，我們就先寫一個這種類型的demo，基於我們前面提供的客戶端和服務端。

　　Server Console

　　這里我提供了2種也是最常用的2種服務端方式，並且提供了不同的處理方式。

class Program
    {
        private static OctMQServer _server;
        static ServerType _type;
        static void Main(string[] args)
        {
            AppDomain.CurrentDomain.UnhandledException += CurrentDomain_UnhandledException;
            CreateCmd();

        }

        /// <summary>
        /// 創建mq對象
        /// </summary>
        static void Create()
        {
            _server = new OctMQServer();
            _server.OnReceive += server_OnReceive;
            _server.Init(5555, _type);
           
        }

        /// <summary>
        /// 選擇類型
        /// </summary>
        private static void CreateCmd()
        {
            Csl.Wl(ConsoleColor.Red, "請選擇您要創建的MQ服務端類型");
            Csl.Wl(ConsoleColor.Yellow, "1.PUB   2.REP");
            var key = System.Console.ReadLine();
            switch (key)
            {
                case "1":
                    {
                        _type = ServerType.Pub;
                        Create();
                        Cmd();
                    }

                    break;
                case "2":
                    _type = ServerType.Response;
                    Create();
                    Cmd();
                    break;
                default:
                    {
                        CreateCmd();
                       
                    }
                    break;
            }
        }

        static void CurrentDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e)
        {
            Csl.WlEx((Exception)e.ExceptionObject);
        }

        /// <summary>
        /// 接收消息
        /// </summary>
        private static void Cmd()
        {
            if (_type == ServerType.Pub)
            {
                Csl.Wl(ConsoleColor.Red, "請輸入您要發個訂閱者的信息主題與信息用空格分開");
            }
            else
            {
                Csl.Wl(ConsoleColor.Red, "等待消息");
            }
            var cmd = System.Console.ReadLine();

            switch (cmd)
            {
                case "exit":
                    Csl.Wl("正在關閉應用程序。。。等待最后一個心跳執行完成。。。");
                    _server.Dispose();
                    break;

                default:
                    {
                        var str = cmd.Split(' ');
                        var msg = _server.CreateMessage();
                        msg.Append(str[0]，Encoding.UTF8);
                        msg.Append(str[1]，Encoding.UTF8);
                        _server.Send(msg);
                        Cmd();
                        break;
                    }
                    return;
            }
        }

        static void server_OnReceive(object sender, DataEventArgs<NetMQ.NetMQSocket, NetMQ.NetMQMessage> e)
        {
            var msg = e.Arg2;
            var server = e.Arg1;
            Csl.Wl(msg.Pop().ConvertToString(Encoding.UTF8));
            server.Send("你好,您的請求已處理，並返回消息及處理結果"，Encoding.UTF8);
        }
    }

　　Client Form 

　　客戶端，我使用winform來處理，並且配合控制台使用，這個用法有些巧妙，不會的同學可以私密我，嘿嘿，先上截圖，也是可以同時處理兩種方式，給個demo，方便大家在實際項目中使用：

　　響應式：

　　

　　訂閱者式：

　　

　　不會做gif ,我就逐步說吧，從訂閱者模式中我們可以看到，我的打開順序1-》2->3,先打開1，訂閱了t的主題，發了2個消息，內容1和內容2，第一個程序均收到，這時我啟動另外一個程序，同樣訂閱t這個主題，發現消息是通過輪詢的方式分別向兩個訂閱者發送，這樣，我們在處理一些比較耗時的業務邏輯，並且不會因為並發出現問題時，就可以使用多個訂閱者，分別處理業務從而大大的提高我們的系統性能。

　　然后打開第三個，訂閱y這個主題，這時發送y的主題消息，前2個訂閱者就無法收到了，這樣我們還可以區分業務，進行多進程的處理，更高的提高可用性和可擴展性，並結合高性能的緩存解決方案處理高並發的業務邏輯。

　　貼出客戶端代碼：

　　

public partial class Form1 : Form
    {
        public Form1()
        {
            InitializeComponent();
            Csl.Init();
        }

        /// <summary>
        /// mq客戶端
        /// </summary>
        private OctMQClient _client;

        /// <summary>
        /// 訂閱者模式連接
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnConn_Click(object sender, EventArgs e)
        {
            _client = new OctMQClient();
            _client.OnReceive += _client_OnReceive;

            _client.Init(txtip.Text,int.Parse(txtport.Text),ClientType.Sub);
            var sub = (SubscriberSocket) _client.Client;
            sub.Subscribe(txtTop.Text);
            _client.StartAsyncReceive();
            
        }

        /// <summary>
        /// 訂閱者模式受到消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        void _client_OnReceive(object sender, Core.Args.DataEventArgs<NetMQ.NetMQSocket, NetMQ.NetMQMessage> e)
        {
            var msg = e.Arg2;
            Csl.Wl("主題："+msg.Pop().ConvertToString(Encoding.UTF8));
            Csl.Wl("內容：" + msg.Pop().ConvertToString(Encoding.UTF8));
        }

        /// <summary>
        /// 發送響應消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btnSend_Click(object sender, EventArgs e)
        {
            using (_client = new OctMQClient())
            {
                _client.Init(txtip.Text, int.Parse(txtport.Text), ClientType.Request);
                var content = txtContent.Text;
                var msg = _client.CreateMessage();
                msg.Append(content, Encoding.UTF8);
                _client.Send(msg);
                var rmsg = _client.ReceiveMessage();
                var reqStr = rmsg.Pop().ConvertToString(Encoding.UTF8);
                Csl.Wl(reqStr);
            }
            
        }

        /// <summary>
        /// 釋放資源
        /// </summary>
        /// <param name="e"></param>
        protected override void OnClosed(EventArgs e)
        {
            base.OnClosed(e);
            if (_client != null)
            {
                _client.Dispose();                
            }
        }
    }

　　好了，大家先消化一下，等系列寫完了，我會提交到github上。下一期，會寫一些並發情況下的應用。