- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在尝试着手控制 TPL 数据流中的数据流。我有一个非常快的生产者和一个非常慢的消费者。 (我的真实代码更复杂,但无论如何,这是一个非常好的模型,它重现了问题。)
当我运行它时,代码开始像过时一样消耗内存——生产者的输出队列尽可能快地填满。我真正希望看到的是生产者停止运行一段时间,直到消费者有机会请求它。根据我对文档的阅读,这是应该发生的情况:也就是说,我认为生产者会等到消费者有空间。
显然,情况并非如此。我该如何解决才能使队列不至于乱成一团?
using System;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Threading;
namespace MemoryLeakTestCase
{
class Program
{
static void Main(string[] args)
{
var CreateData = new TransformManyBlock<int, string>(ignore =>
{
return Enumerable.Range(0, 1000 * 1000 * 1000).Select((s,i) => "Hello, World " + i);
});
var ParseFile = new TransformManyBlock<string, string>(fileContent =>
{
Thread.Sleep(1000);
return Enumerable.Range(0, 100).Select((sst, iii) => "Hello, " + iii);
}, new ExecutionDataflowBlockOptions() { BoundedCapacity = 1000 }
);
var EndOfTheLine = new ActionBlock<object>(f =>
{
});
var linkOptions = new DataflowLinkOptions { PropagateCompletion = true, };
CreateData.LinkTo(ParseFile, linkOptions);
ParseFile.LinkTo(EndOfTheLine, linkOptions);
Task t = new Task(() =>
{
while (true)
{
Console.WriteLine("CreateData: " + Report(CreateData));
Console.WriteLine("ParseData: " + Report(ParseFile));
Console.WriteLine("NullTarget: " + EndOfTheLine.InputCount );
Thread.Sleep(1000);
}
});
t.Start();
CreateData.SendAsync(0);
CreateData.Complete();
EndOfTheLine.Completion.Wait();
}
public static string Report<T, U>(TransformManyBlock<T, U> block)
{
return String.Format("INPUT: {0} OUTPUT: {1} ", block.InputCount.ToString().PadLeft(10, ' '), block.OutputCount.ToString().PadLeft(10, ' '));
}
}
}
最佳答案
通常,在这种情况下,您会做的是同时设置 CreateData
block 的 BoundedCapacity
。但这在这里行不通,因为在从单个 IEnumerable
填充输出队列时,TransformManyBlock
似乎没有考虑 BoundedCapacity
。
您可以改为创建一个函数来迭代集合并使用 SendAsync()
仅当目标可以接受它们时才发送更多数据:
/// <remarks>
/// If iterating data throws an exception, the target block is faulted
/// and the returned Task completes successfully.
///
/// Depending on the usage, this might or might not be what you want.
/// </remarks>
public static async Task SendAllAsync<T>(
this ITargetBlock<T> target, IEnumerable<T> data)
{
try
{
foreach (var item in data)
{
await target.SendAsync(item);
}
}
catch (Exception e)
{
target.Fault(e);
}
}
用法:
var data = Enumerable.Range(0, 1000 * 1000 * 1000).Select((s,i) => "Hello, World " + i);
await ParseFile.SendAllAsync(data);
ParseFile.Complete();
如果您仍然希望 CreateData
block 的行为与您的原始代码类似,您可以有两个有界的 BufferBlock
,SendAllAsync()
然后使用 Encapsulate()
使它们看起来像一个 block :
/// <remarks>
/// boundedCapacity represents the capacity of the input queue
/// and the output queue separately, not their total.
/// </remarks>
public static IPropagatorBlock<TInput, TOutput>
CreateBoundedTransformManyBlock<TInput, TOutput>(
Func<TInput, IEnumerable<TOutput>> transform, int boundedCapacity)
{
var input = new BufferBlock<TInput>(
new DataflowBlockOptions { BoundedCapacity = boundedCapacity });
var output = new BufferBlock<TOutput>(
new DataflowBlockOptions { BoundedCapacity = boundedCapacity });
Task.Run(
async () =>
{
try
{
while (await input.OutputAvailableAsync())
{
var data = transform(await input.ReceiveAsync());
await output.SendAllAsync(data);
}
output.Complete();
}
catch (Exception e)
{
((IDataflowBlock)input).Fault(e);
((IDataflowBlock)output).Fault(e);
}
});
return DataflowBlock.Encapsulate(input, output);
}
关于c# - 如何安排 TPL 数据流中的流控制?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/20715816/
我们有一个接口服务为下游的系统提供数据服务,本来好好的大家都愉快地传递JSON,非常和谐。可最近有个新需求去对接一个很老的系统,这倒是不算啥,可这个老系统数据不是以JSON传递的而是以XML传递
我想完成这样的事情: results = [] for i in range(N): data = generate_data_slowly() res = tasks.process
如何安排 SSIS 每天在给定时间从文件中自动执行包? 最佳答案 你有几个选择,但我会给你一些让你快速启动和运行的东西...... 打开 SQL Server Management Studio 并连
我们有两个服务器 Azure 配置,运行由 Umbraco 提供支持的网站。当我们需要向Azure服务器添加新域时,我们修改配置文件,然后依次重新启动每台服务器。 理论上,因为我们永远不会同时重新启动
我给出了一个创建电子邮件 C# 控制台应用程序的任务,该应用程序的目标是批量运行。我对 C# 领域非常陌生,因此我不知道我的方向。此 C# 控制台应用程序将部署在服务器上,并期望根据服务器时间在特定时
我有一个控制台应用程序,运行时会执行一些操作,并使用docker生成它的镜像。现在,我想将其部署到Kubernetes并每小时运行一次,是否有可能在K8中完成? 我已经阅读了有关Cron作业的信息,但
这是我的 CronJob 规范的一部分: kind: CronJob spec: schedule: #{service.schedule} 对于特定环境,设置了 cron 作业,但我从不希望
我的任务是创建一个应用程序,该应用程序将每 (n) 分钟向选定的收件人发送一封电子邮件。它所在的应用程序的结构方式是通过回调 .main(args) 来重置自身。每当需要的时候。我的问题是,当我调用.
安排 Airflow Dag 使其仅在工作日运行的正确方法是什么?我已经尝试在 start_date 和 schedule_interval 表达式中都包含小时偏移量,但它仍然没有在所需的时间开始。
我有许多测试都安排了一些 TestFixtures,我发现我正在复制该安排代码很多。每个测试的前几行几乎相同。 有没有一种方法可以在所有测试中声明一个共享的 TestFixture,同时仍然在每个测试
我有一个问题,我正在创建一个应用程序,我想在系统与 azan 时间匹配时在后台播放 azan 文件,无论用户正在使用应用程序的任何屏幕,azan 都应该开始播放。 我在 Azan.java 中创建了一
在我没有重启我的手机之前一直在 toast ,但是在重启之后 broadcastreceiver2 没有收到并且没有任何反应。 我关注了http://stacktips.com/tutorials/a
自动将一个数据库表的表数据复制到另一个数据库表;当表格更新或按某个特定时间间隔更新时,安排 数据库MySQL;语言 PHP 我有两个数据库; A和B 数据库 A 包含一个表 USERS 我想将USER
我的 Android 应用程序将定期轮询服务器以检查数据。我希望无论用户与应用程序交互如何进行此轮询,类似于(在概念上)Gmail 和 Google Reader 应用程序如何在后台同步数据。安装应用
我可以将android中的警报管理器(.set()方法)安排到当前时间一个月后的时间吗它会活那么久吗?操作系统对此 alarmManager 有何影响? 最佳答案 用户重启手机时的提示。您可以使用以下
安排 AsyncTask 每分钟运行一次的最佳做法是什么(请注意,在 AsyncTask 完成后我应该能够更新 UI)。 我不打算使用服务,因为这些任务应该只在应用处于 Activity 状态时运行。
我在排列从 php 中的 while 循环返回的数据时遇到问题。 基本上,我正在尝试从数据库返回工作的时间段计划,问题是我似乎在所有时间段中得到相同的结果,或者在一个时间段中的所有客户端得到相同的结果
我想创建一个仅在周六和周四运行的 mysql 事件。 是否可以定义事件本身的日期? 我有一个想法,每天运行调度程序,如果是星期四或星期六,则该过程将继续,否则它将退出调度程序而不执行任何操作。 最佳答
如何使用 MySQL 调度程序安排查询运行(如果这是最好的方法)?我按照 link here 中的说明进行操作但我有点迷路了。 我想在我们拥有的特定数据库上每 30 分钟运行一次以下查询。 u
我想在使用事件轮换我的日志后读取我的表日志,我希望我的事件在我选择的一周中的任何一天运行。 经过一番研究,我想到了这个 CREATE EVENT read_rotated_logs ON SCHEDU
我是一名优秀的程序员,十分优秀!