Reactor模式的.net版本簡單實現--DEMO

     近期在學習DotNetty，遇到不少的問題。由於dotnetty是次netty的.net版本的實現。導致在網上敘述dotnetty的原理，以及實現技巧方面的東西較少，這還是十分惱人的。在此建議學習和使用Dotnetty的和位小伙伴，真心閱讀下netty的相關書籍，如《netty權威指南》。

     閑話少說，進入正題。netty的性能之所以能夠達到如此的高度。主要由於他使用Reactor模式處理socket的請求，讓伺服器的使用率最大化，且儘量減少線程的開銷。本文章主要簡單介紹下Reactor模式。

一、reactor概論

reactor模式主要解決處理多個客戶端請求的設計模式。

首先從類圖我們可以得知：

Dispatcher：Handler管理器，以及調用度。他依賴於Demultiplexer類

Demultiplexer：事件管理器，接受外部的事件，並提供給Dispatch使用。

Handle：事件源，表示觸發了那些事件

EventHandler：各種類型的處理器，用於處理具體的業務，以及I/O的讀寫

當然，也可以通過序列圖看出首先需要初始化Dispatcher, Demultiplexer等相關類，以及註冊具體的事件處理器。

二、代碼的具體實現

類圖如下[源碼下載]：

2.1 多路復用事件處理器的代碼

public class Demultiplexer
    {
        private ConcurrentQueue<Event> eventQuene = new ConcurrentQueue<Event>();
        private Object lockObj = new Object();

        public List<Event> Select()
        {
            return this.Select(0);
        }
        public List<Event> Select(int time)
        {
            if(time > 0)
            {
                if (this.eventQuene.IsEmpty)
                {
                    lock (lockObj)
                    {
                        if (this.eventQuene.IsEmpty)
                        {
                            System.Threading.Thread.Sleep(time);
                        }
                    }
                }
            }
            List<Event> events = new List<Event>();
            while(this.eventQuene.Count > 0)
            {
                Event tmp;
                if(this.eventQuene.TryDequeue(out tmp))
                {
                    events.Add(tmp);
                }
            }
            return events;
        }
        public void AddEvent(Event argEvent)
        {
            this.eventQuene.Enqueue(argEvent);
        }
    }

此類主要防止多線程的共同競爭，因為多路徑復用選擇器會被多個線程同時使用。所以使用的線程安全的Queue。

2.2 Handler觸發器和管理器

/// <summary>
    /// Reactor的事件Handler觸發器，提供事件Handler的註冊，移除
    /// </summary>
    public class EventDispatch
    {
        private Demultiplexer demultiplexer;
        Dictionary<EventType, EventHandler> eventHandlerMap = new Dictionary<EventType, EventHandler>();

        public EventDispatch(Demultiplexer demultiplexer)
        {
            this.demultiplexer = demultiplexer;
        }

        public void RegisterHandler(EventType eventType, EventHandler eventHandler)
        {
            this.eventHandlerMap.Add(eventType, eventHandler);
        }
        public void RemoveHandler(EventType eventType)
        {
            this.eventHandlerMap.Remove(eventType);
        }

        public void HandleEvents()
        {
            this.Dispatch();
        }
        public void Dispatch()
        {
            string log = string.Format("thread id: {0} Dispatch", System.Threading.Thread.CurrentThread.ManagedThreadId);
            Console.WriteLine(log);
            while (true)
            {
                List<Event> events = this.demultiplexer.Select();
                foreach(var itemEvent in events)
                {
                    EventHandler eventHandler = this.eventHandlerMap[itemEvent.EventType];
                    eventHandler.Handle(itemEvent);
                }
                System.Threading.Thread.Sleep(1000);
            }
        }
    }

主要職責，對Handler的註冊、移除的管理，以及通過 多路復用選擇器 選擇相應的Handler進行處理。

2.3 服務端的實現

/// <summary>
    /// 開啟接受請求的服務端
    /// </summary>
    public class AcceptRuner
    {
        private System.Collections.Concurrent.ConcurrentQueue<object> sourceQueue = new System.Collections.Concurrent.ConcurrentQueue<object>();

        private Demultiplexer demultiplexer;

        public AcceptRuner(Demultiplexer demultiplexer)
        {
            this.demultiplexer = demultiplexer;
        }

        public void adConnection(object source)
        {
            this.sourceQueue.Enqueue(source);
        }

        public void Run()
        {
            string log = string.Format("thread id: {0} AcceptRunner", System.Threading.Thread.CurrentThread.ManagedThreadId);
            Console.WriteLine(log);
            while (true)
            {
                object source;
                if(this.sourceQueue.TryDequeue(out source))
                {
                    Event acceptEvent = new Event()
                    {
                        EventType = EventType.Accept,
                        Source = source
                    };
                    this.demultiplexer.AddEvent(acceptEvent);
                }
            }
        }
    }

此類效仿netty的serverBoostrap的實現，將外部新的連接以事件對象的形式添加到 多路復用選擇器上。

2.4 其他類

Event：事件基類

EventHandler：事件處理器抽象基類。他派生了：AcceptEventHandler，ReadEventHandler。

EventType：事件類型

三、備註說明

1. 代碼沒有貼完整。但下載包就是完整的。

2. 這隻我對Reactor模式的理解，如有偏頗之處，還望各拉指點一二。