(一)TDF中的负载均衡 【代码地址-Blance1】
我们可以使用Block很方便的构成一个生产者消费者的模式来处理数据。当生产者产生数据的速度快于消费者的时候,消费者Block的Buffer中的数据会越来越多,消耗大量的内存,数据处理也会延时。这时,我们可以用一个生产者Block连接多个消费者Block来解决这个问题。由于多个消费者Block一定是并行处理,所以对共享资源的处理一定要做同步处理。
使用BoundedCapacity属性来实现
当连接多个ActionBlock的时候,可以通过设置ActionBlock的BoundedCapacity属性。当第一个满了,就会放到第二个,第二个满了就会放到第三个。
public static BufferBlock<string> bb = new BufferBlock<string>();private const string DateFormat = "yyyy-MM-dd HH:mm:ss.fff";public Form1(){InitializeComponent();bb.LinkTo(ab1);bb.LinkTo(ab2);bb.LinkTo(ab3);for (int i = 1; i <= 20; i++){Thread.Sleep(10);string str = $"【BufferBlock Post:{DateTime.Now.ToString(DateFormat)} 值:{i} bb长度:{bb.Count}】";bb.Post(str);}}public static ActionBlock<string> ab1 = new ActionBlock<string>((i) =>{Thread.Sleep(1000);Console.WriteLine($"【ab1】{DateTime.Now.ToString(DateFormat)} {i} ab1长度:{ab1.InputCount} bb长度:{bb.Count}");}, new ExecutionDataflowBlockOptions() { BoundedCapacity = 2 });public static ActionBlock<string> ab2 = new ActionBlock<string>((i) =>{Thread.Sleep(2000);Console.WriteLine($"【ab2】{DateTime.Now.ToString(DateFormat)} {i} ab2长度:{ab2.InputCount} bb长度:{bb.Count}");}, new ExecutionDataflowBlockOptions() { BoundedCapacity = 2 });public static ActionBlock<string> ab3 = new ActionBlock<string>((i) =>{Thread.Sleep(3000);Console.WriteLine($"【ab3】{DateTime.Now.ToString(DateFormat)} {i} ab3长度:{ab3.InputCount} bb长度:{bb.Count}");}, new ExecutionDataflowBlockOptions() { BoundedCapacity = 2 });
分析以上数据可知: