爲Disruptor 寫的一個簡單實用的.Net擴展

 

disruptor   用戶封裝本身的消費者,把消費者注入到消費者容器,消費者容器實現自動建立 緩存隊列,生產者;

 文中用到的 disruptor   C#移植源代碼

 https://github.com/bingyang001/disruptor-net-3.3.0-alpha

 做者博客 http://www.cnblogs.com/liguo/p/3296166.html

 

 消費者容器:

/// <summary>
    /// 消費者管理器
    /// </summary>
    /// <typeparam name="TProduct">產品</typeparam>
    public class Workers<TProduct> where TProduct : Producer<TProduct>, new()
    {
        private readonly WorkerPool<TProduct> _workerPool;

        public Workers(List<IWorkHandler<TProduct>> handers, IWaitStrategy waitStrategy = null, int bufferSize = 1024*64)
        {
            if (handers == null || handers.Count == 0)
                throw new ArgumentNullException("消費事件處理數組爲空!");
            if (handers.Count == 1)
                _ringBuffer = RingBuffer<TProduct>.CreateSingleProducer(() => new TProduct(), bufferSize,
                    waitStrategy ?? new YieldingWaitStrategy());
            else
            {
                _ringBuffer = RingBuffer<TProduct>.CreateMultiProducer(() => new TProduct(), bufferSize,
                    waitStrategy ?? new YieldingWaitStrategy());
            }
            _workerPool = new WorkerPool<TProduct>(_ringBuffer
                , _ringBuffer.NewBarrier()
                , new FatalExceptionHandler()
                , handers.ToArray());
            _ringBuffer.AddGatingSequences(_workerPool.getWorkerSequences());
        }

        public void Start()
        {
            _workerPool.start(TaskScheduler.Default);
        }

        public Producer<TProduct> CreateOneProducer()
        {
            return new Producer<TProduct>(this._ringBuffer);
        } 
        public void DrainAndHalt()
        {
            _workerPool.drainAndHalt();
        }

        private readonly RingBuffer<TProduct> _ringBuffer;
    }

 

  生產者(產品): 全部的產品都應該繼承自生產者

/// <summary>
    /// 生產者對象
    /// </summary>
    /// <typeparam name="TProduct">產品類型</typeparam>
    public class Producer<TProduct> where TProduct:Producer<TProduct>
    {

        long _sequence;
        private RingBuffer<TProduct> _ringBuffer;
        public Producer()
        {
            
        }
        public Producer(RingBuffer<TProduct> ringBuffer )
        {
            _ringBuffer = ringBuffer;
        }
        /// <summary>
        /// 獲取可修改的產品
        /// </summary>
        /// <returns></returns>
        public Producer<TProduct> Enqueue()
        {
            long sequence = _ringBuffer.Next();
            Producer<TProduct> producer = _ringBuffer[sequence];
            producer._sequence = sequence;
            if (producer._ringBuffer == null)
                producer._ringBuffer = _ringBuffer;
            return producer;
        }
        /// <summary>
        /// 提交產品修改
        /// </summary>
        public void Commit()
        {
            _ringBuffer.Publish(_sequence);
        }
    }

 

  --------------------------------------------------------

  以上就實現了,測試代碼

  先建立 產品對象:

  

/// <summary>
        /// 產品/繼承生產者
        /// </summary>
        public class Product : Producer<Product>
        {
            //產品包含的屬下隨便定義,無要求,只須要繼承自生產者就好了
            public long Value { get; set; }
            public string Guid { get; set; }
        }

建立消費者對象

 /// <summary>
        /// 消費處理對象
        /// </summary>
        public class WorkHandler : IWorkHandler<Product>
        {
         
            public void OnEvent(Product @event)
            {
                //Test是測試對象數據準確(數據重複或者丟失數據)
                Test.UpdateCacheByOut(@event.Guid);
                //收到產品,在這裏寫處理代碼

            }

        }

  測試代碼:

  可建立1個或者多個的生產者對象,消費者處理對象;不必定太多,多不必定快; 建議生產者建立一個就好了,多線程操做一個生產者對象; 消費者對象能夠根據實際狀況建立多少個;

  

           //建立2個消費者,2個生產者, 2個消費者表示,框架會有2個線程去處理消費產品 
           Workers<Product> workers = new Workers<Product>(
            new List<IWorkHandler<Product>>() {new WorkHandler(), new WorkHandler()});

            Producer<Product> producerWorkers = workers.CreateOneProducer();
            Producer<Product> producerWorkers1 = workers.CreateOneProducer();
           //開始消費
             workers.Start();

 產品生產:

 能夠在任何引用生產者的地方,把產品放進隊列中. 這裏 放入隊列的方法和平時不太同樣.  這裏採用的是,從隊列裏面拿去一個位置,而後把產品放進去; 具體的作法 ,找生產者,獲取一個產品對象,而後修改產品屬性,最後提交修改.

  var obj = producer.Enqueue();
           //修改產品屬性
                obj.Commit();

 

  以上是關鍵代碼:

完整的測試類 : 包含測試數據正確性,  性能,在不校驗正確性的時候,每秒ops 1千萬左右. 

 class Test
    {
        public static long PrePkgInCount = 0;
        public static long PrePkgOutCount = 0;
        public static long PkgInCount = 0;
        public static long PkgOutCount = 0;
        static ConcurrentDictionary<string, string> InCache = new ConcurrentDictionary<string, string>();
        static ConcurrentDictionary<string, string> OutCache = new ConcurrentDictionary<string, string>();
        private static long Seconds;

        static void Main(string[] args)
        {
            Workers<Product> workers = new Workers<Product>(
            new List<IWorkHandler<Product>>() {new WorkHandler(), new WorkHandler()});

            Producer<Product> producerWorkers = workers.CreateOneProducer();
            Producer<Product> producerWorkers1 = workers.CreateOneProducer();

            workers.Start();
            Task.Run(delegate
            {
                while (true)
                {
                    Thread.Sleep(1000);
                    Seconds++;
                    long intemp = PkgInCount;
                    long outemp = PkgOutCount;
                    Console.WriteLine(
                        $"In ops={intemp - PrePkgInCount},out ops={outemp - PrePkgOutCount},inCacheCount={InCache.Count},OutCacheCount={OutCache.Count},RunningTime={Seconds}");
                    PrePkgInCount = intemp;
                    PrePkgOutCount = outemp;
                }

            });
            Task.Run(delegate { Run(producerWorkers); });
            Task.Run(delegate { Run(producerWorkers); });
            Task.Run(delegate { Run(producerWorkers1); });
            Console.Read();

        }

        public static void Run(Producer<Product> producer)
        {
            for (int i = 0; i < int.MaxValue; i++)
            {

                var obj = producer.Enqueue();
                CheckRelease(obj as Product);
                obj.Commit();
            }
        }

        public static  void CheckRelease(Product publisher)
        {
            Interlocked.Increment(ref PkgInCount);
            return; //不檢查正確性
            publisher.Guid = Guid.NewGuid().ToString();
            InCache.TryAdd(publisher.Guid, string.Empty);
          
        }

        public static void UpdateCacheByOut(string guid)
        {
            Interlocked.Increment(ref Test.PkgOutCount);
            if (guid != null)
                if (InCache.ContainsKey(guid))
                {
                    string str;
                    InCache.TryRemove(guid, out str);
                }
                else
                {
                    OutCache.TryAdd(guid, string.Empty);
                }

        }
        /// <summary>
        /// 產品/繼承生產者
        /// </summary>
        public class Product : Producer<Product>
        {
            //產品包含的屬下隨便定義,無要求,只須要繼承自生產者就好了
            public long Value { get; set; }
            public string Guid { get; set; }
        }

        /// <summary>
        /// 消費處理對象
        /// </summary>
        public class WorkHandler : IWorkHandler<Product>
        {
         
            public void OnEvent(Product @event)
            {

                Test.UpdateCacheByOut(@event.Guid);
                //收到產品,在這裏寫處理代碼

            }

        }
    }