C# DataFlow入门（一、九大Block）

 #region 九大block

        #region 一、bufferblock
        
        private BufferBlock m_buffer = new BufferBlock();
        private void Producer() {
            Random random = new Random();
            while (true) {
                int item = random.Next(10, 50);
                Console.WriteLine("P:" + item);
                m_buffer.Post(item);
                Thread.Sleep(100);
            }
        }
        private void Consumer() {
            while (true)
            {
                int item = m_buffer.Receive();
                Console.WriteLine("C:" + item);
                Thread.Sleep(100);
            }
        }

        public void TestBufferBlock() {
            var p = Task.Factory.StartNew(Producer);
            var c = Task.Factory.StartNew(Consumer);
            Task.WaitAll(p, c);
        }
        #endregion

        #region 二、ActionBlock
        
        private ActionBlock m_action = new ActionBlock((i) => {
            Thread.Sleep(1000);
            Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
        }, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism=3});

        private ActionBlock> m_action2 = new ActionBlock>((i) => {
            Thread.Sleep(1000);
            Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
        }, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 3 });

        public void TestActionBlock()
        {
            var p = Task.Factory.StartNew(()=> {
                for (int i = 0; i < 10; i++)
                {
                    m_action.Post(i);
                    if (i == 8) {
                        m_action.Complete(); // 控制生命周期
                    }
                }
                
                Console.WriteLine("Post finished");
            });
            Task.WaitAll();
            m_action.Completion.Wait(); // 等待ActionBlock处理结束
            Console.WriteLine("m_action.Completion finished");
        }

        public void TestActionBlock2()
        {
            var p = Task.Factory.StartNew(() => {
                for (int i = 0; i < 10; i++)
                {
                    m_action.Post(i);
                    if (i == 8)
                    {
                        m_action.Complete(); // 控制生命周期
                    }
                }

                Console.WriteLine("Post finished");
            });

            Task.WaitAll();
            m_action.Completion.Wait();
            Console.WriteLine("m_action.Completion finished");
        }
        #endregion

        #region 三、TransformBlock
        
        private TransformBlock tbUrl = new TransformBlock((i) =>
        {
            Thread.Sleep(500);
            Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
            return i;
        });

        public void TestTransformBlock()
        {
            for (int i = 0; i < 10; i++) {
                tbUrl.Post(i);
            }

            while (!tbUrl.Completion.IsCompleted) {
                int i = tbUrl.Receive();
                Console.WriteLine(i+"处理完成");
            }
            
        }
        #endregion

        #region 四、TransformManyBlock
        
        TransformManyBlock tmb = new TransformManyBlock((i) => { return new int[] { i, i + 1 }; });

        ActionBlock ab = new ActionBlock((i) => Console.WriteLine(i));

        public void TestTransformManyBlock()
        {
            tmb.LinkTo(ab);

            for (int i = 0; i < 4; i++)
            {
                tmb.Post(i);
            }

            Console.WriteLine("Finished post");
        }
        #endregion

        #region 五、BroadcastBlock
        
        BroadcastBlock bb = new BroadcastBlock((i) => { return i; });

        ActionBlock displayBlock = new ActionBlock((i) => Console.WriteLine("Displayed " + i));

        ActionBlock saveBlock = new ActionBlock((i) => Console.WriteLine("Saved " + i));

        ActionBlock sendBlock = new ActionBlock((i) => Console.WriteLine("Sent " + i));

        public void TestBroadcastBlock()
        {
            bb.LinkTo(displayBlock);
            bb.LinkTo(saveBlock);
            bb.LinkTo(sendBlock);

            for (int i = 0; i < 4; i++)
            {
                bb.Post(i);
            }

            Console.WriteLine("Post finished");
        }
        #endregion

        #region 六、WriteOnceBlock
        
        WriteOnceBlock wb = new WriteOnceBlock((i) => { return i; });

        ActionBlock displayBlock2 = new ActionBlock((i) => Console.WriteLine("Displayed " + i));

        ActionBlock saveBlock2 = new ActionBlock((i) => Console.WriteLine("Saved " + i));

        ActionBlock sendBlock2 = new ActionBlock((i) => Console.WriteLine("Sent " + i));

        public void TestWriteOnceBlock()
        {
            wb.LinkTo(displayBlock2);
            wb.LinkTo(saveBlock2);
            wb.LinkTo(sendBlock2);
            for (int i = 0; i < 4; i++)
            {
                wb.Post(i);
            }

            Console.WriteLine("Post finished");
        }
        #endregion

        #region 七、BatchBlock
        
        BatchBlock bb1 = new BatchBlock(3);

        ActionBlock ab1 = new ActionBlock((i) =>
        {
            string s = string.Empty;

            foreach (int m in i)
            {
                s += m + " ";
            }
            Console.WriteLine(s);
        });

        public void TestBatchBlock()
        {
            bb1.LinkTo(ab1);

            for (int i = 0; i < 10; i++)
            {
                bb1.Post(i);
            }
            bb1.Complete();

            Console.WriteLine("Finished post");
        }
        #endregion

        #region 八、JoinBlock
        
        JoinBlock jb = new JoinBlock();
        ActionBlock> ab2 = new ActionBlock>((i) =>
        {
            Console.WriteLine(i.Item1 + " " + i.Item2);
        });

        public void TestJoinBlock()
        {
            jb.LinkTo(ab2);

            for (int i = 0; i < 5; i++)
            {
                jb.Target1.Post(i);
            }

            for (int i = 5; i > 0; i--)
            {
                Thread.Sleep(1000);
                jb.Target2.Post(i.ToString());
            }

            Console.WriteLine("Finished post");
        }
        #endregion

        #region 九、BatchedJoinBlock
        
        BatchedJoinBlock bjb = new BatchedJoinBlock(3);

        ActionBlock, IList>> ab3 = new ActionBlock, IList>>((i) =>
        {
            Console.WriteLine("-----------------------------");

            foreach (int m in i.Item1)
            {
                Console.WriteLine(m);
            };

            foreach (string s in i.Item2)
            {
                Console.WriteLine(s);
            };
        });

        public void TestBatchedJoinBlock()
        {
            bjb.LinkTo(ab3);

            for (int i = 0; i < 5; i++)
            {
                bjb.Target1.Post(i);
            }

            for (int i = 5; i > 0; i--)
            {
                bjb.Target2.Post(i.ToString());
            }

            Console.WriteLine("Finished post");
        }
        #endregion

        #endregion

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Demo
{
    internal class Program
    {
        static void Main(string[] args)
        {
            TPLDemo demo = new TPLDemo();   
            //demo.TestBufferBlock();
            //demo.TestActionBlock();
            //demo.TestActionBlock2();
            //demo.TestTransformBlock();
            //demo.TestTransformManyBlock();
            //demo.TestBroadcastBlock();
            //demo.TestWriteOnceBlock();
            //demo.TestBatchBlock();
            //demo.TestJoinBlock();
            demo.TestBatchedJoinBlock();
            Console.ReadLine();
        }
    }
}