我正在尝试并行执行大量网络操作,并且我想为每个操作设置一个超时时间。
由于 Parallel.ForEach 没有简单的超时选项,我正在使用 System.Reactive。
这是我的代码:
public void networkOps(List<MacCpe> source, Action<List<Router>, List<Exception>> onDone) {
var routers = new List<Router>();
var exceptions = new List<Exception>();
Observable.Defer(() => source.ToObservable())
.ObserveOn(Scheduler.CurrentThread)
.SubscribeOn(Scheduler.Default)
.SelectMany(it =>
Observable.Amb(
Observable.Start(() => {
switch(it.type) {
case AntennaType.type1: {
//network stuff
}
break;
case AntennaType.type2: {
//network stuff
}
break;
case AntennaType.type3: {
//network stuff
}
break;
case AntennaType.type4: {
//network stuff
}
break;
default: throw new NullReferenceException("Nothing");
}
}).Select(_ => true),
Observable.Timer(TimeSpan.FromSeconds(60)).Select(_ => false)
),
(it, result) => new { it, result }
)
.Subscribe (
x => {
Console.WriteLine("checked item number " + x.it.Id);
},
ex => {
Console.WriteLine("error string");
}, () => {
onDone(routers, exceptions);
}
);
}
我正在使用 Observable.Amb 运算符并行运行一个 60 秒计时器,作为超时。
但是,当我运行此方法时,程序会立即退出,而不会到达回调 onDone。
我在网上看到我可以使用 ObserveOnDispatcher 在线程池上运行阻塞代码时观察 Ui 线程,但我在终端应用程序服务器端的 linux 上的 dotnet 核心上使用它。
如何去观察控制台应用程序中的“主线程”?
预先感谢您的回复。
最佳答案
当您替换 Parallel.ForEach
时,听起来您很乐意进行阻塞操作。按照您设置的方式使用 Rx,它不是阻塞操作,因此该方法会立即结束。
修复起来非常简单。只需将您的 .Subscribe
更改为:
.Do(
x =>
{
Console.WriteLine("checked item number " + x.it.Id);
},
ex =>
{
Console.WriteLine("error string");
}, () =>
{
onDone(routers, exceptions);
}
)
.Wait();
我还会去掉你的 .ObserveOn(Scheduler.CurrentThread)
和 .SubscribeOn(Scheduler.Default)
直到你确定你需要那些。
关于linux - Reactive Extensions .Net Core 在主线程上观察,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51022709/