间歇输入数据的数据处理设计模式

目录

介绍

背景

使用代码

兴趣点

• 下载源1.5 KB

如果您的数据是间歇性的（非连续性），那么我们可以利用时间跨度间隔来优化CPU\RAM利用率。这个想法是在下一批数据到达之前处理数据。假设您每T秒接收N个输入数据，每个数据的大小为d，而一个数据需要P秒的处理时间。如果使用单线程，则所需的总输出时间将为N x P秒。如果N x P T，那么您需要多个线程，即，当处理输入所需的时间大于两个连续批次的数据之间的时间。如果我们为多个线程引入另一个变量，那么我们的问题将简化为 [ (N x P) / c ] < T。

下一个约束是可以创建多少个线程？那限制了因子c。如果c太高，则会消耗大量CPU。在这里，我们引入了RAM利用率。随着数据的传入，我们首先将其存储在内存中，然后使用c线程对其进行处理。因此，在任何时候，队列中都会有c个活动线程和Nc个待处理项目。假设r个可以在内存中的批次，一个批次可以一次由c个线程处理。一批次大小为cxd。现在我们可以将其归结为：

• [ (N x P) / (r x c) ] < T
• r = Affordable RAM / (c x d)

当您以特定的频率收集数据时，此方案主要适用于基于轮询的系统。因此，假设数据流是间歇性的并且间隔一定时间发生。您可以利用数据收集之间的时间间隔来最佳利用CPU和RAM。

我们需要一种调查性的数据处理方法，因为一种尺寸无法容纳所有尺寸。诸如N，d和P之类的许多参数事先未知。因此，我们需要设计还提供统计信息，以便我们可以了解N，d和P并相应地调整CPU和RAM的需求。

作为粗略的指导，我们需要一种方法来摄取通过线程提交的所有数据。然后，立即开始处理它们或将它们排成队列并在多个线程中处理它们。

C＃为线程安全的集合提供了阻止和限制功能。这是一个有趣的功能，可用于优化高负载应用程序的CPU和内存。此模式可以进一步堆叠和互连，以构建数据路由的有向图。该模式已在Apache Nifi处理器中广泛使用。

在深入探讨模式之前，让我们了解什么是边界和阻塞。他们解决什么问题？

当有多个线程试图从容器中获取数据时，我们希望线程阻塞直到有更多数据可用为止。这称为“阻止”。

当多个线程正在写入数据时，我们希望它们进行绑定，直到有一些可用的内存来容纳新数据。这称为“边界”。

因此，我们可以将阻塞集合用作基础数据容器。

BlockingCollection DataContainer = new BlockingCollection(
new ConcurrentBag(),
this.MaxContainerSize);

Thread[] Workers = new Thread[this.MaxWorkerThreads];

for (int i = 0; i < Workers.Length; i++)
{

    Thread newThread = new Thread(new ParameterizedThreadStart(ThreadFunction));
    Workers[i] = newThread;
}

private void ThreadFunction(object threadContext)
{
       CancellationToken token = (CancellationToken)threadContext;
       while (!token.IsCancellationRequested)
       {
              string Data = DataContainer.Take();
              ProcessData(Data);           
       }
}

public void Add(string data)
{
    DataContainer.Add(data);
}

现在要优化和调整RAM和CPU利用率，您需要调整MaxWorkerThreads和MaxContainerSize。我们需要收集一些统计信息以了解数据流模式。

1. 输入速率或每秒有多少数据？
2. 输出速率或每秒处理多少数据？
3. 平均活动线程
4. 平均容器尺寸

这些指标通过以下方式提供帮助：

1. 如果“输入速率”>“输出速率”，则容器大小将永远增长，或者输入处的阻塞线程将增加，但会使程序崩溃。

•                         所以输入速率