反应式扩展:如何观察 IEnumerable 方法结果异步
本文关键字:IEnumerable 方法 结果 异步 观察 扩展 何观察 反应式 | 更新日期: 2023-09-27 18:37:00
我有一个返回我的业务对象的 IEnumerable 的方法。在此方法中,我将大型文本文件的内容解析为业务对象模型。里面没有线程的东西。
在我的视图模型 (WPF) 中,我需要存储和显示方法的结果。存储是一个可观察集合。
这里是可观察的代码:
private void OpenFile(string file)
{
_parser = new IhvParser();
App.Messenger.NotifyColleagues(Actions.ReportContentInfo, new Model.StatusInfoDisplayDTO { Information = "Lade Daten...", Interval = 0 });
_ihvDataList.Clear();
var obs = _parser.ParseDataObservable(file)
.ToObservable(NewThreadScheduler.Default)
.ObserveOnDispatcher()
.Subscribe<Ihv>(AddIhvToList, ReportError, ReportComplete);
}
private void ReportComplete()
{
App.Messenger.NotifyColleagues(Actions.ReportContentInfo, new Model.StatusInfoDisplayDTO { Information = "Daten fertig geladen.", Interval = 3000 });
RaisePropertyChanged(() => IhvDataList);
}
private void ReportError(Exception ex)
{
MessageBox.Show("...");
}
private void AddIhvToList(Ihv ihv)
{
_ihvDataList.Add(ihv);
}
这是解析器代码:
public IEnumerable<Model.Ihv> ParseDataObservable(string file)
{
using (StreamReader reader = new StreamReader(file))
{
var head = reader.ReadLine(); //erste Zeile ist Kopfinformation
if (!head.Contains("BayBAS") || !head.Contains("2.3.0"))
{
_logger.ErrorFormat("Die Datei {0} liegt nicht im BayBAS-Format 2.3.0 vor.");
}
else
{
while (!reader.EndOfStream)
{
var line = reader.ReadLine();
if (line.Length != 1415)
{
_logger.ErrorFormat("Die Datei {0} liegt nicht im BayBAS-Format 2.3.0 vor.");
break;
}
var tempIhvItem = Model.Ihv.Parse(line);
yield return tempIhvItem;
}
reader.Close();
}
}
}
为什么我没有异步获取结果?在 DataGrid 中看到结果之前,将解析并传递所有项目。
有人可以帮忙吗?
安德烈亚斯
你确定这不是异步发生的吗? 您是根据您在 UI 中感知到的内容假设这一点,还是设置了断点并确定事实确实如此?
请注意,WPF 的Dispatcher
使用优先级队列,DispatcherScheduler
以Normal
优先级计划项,优先级高于用于输入、布局和呈现的优先级。 如果结果来得足够快,则 UI 可能要等到处理完最后一个结果后才会更新:调度程序可能忙于处理结果,无法执行 UI 的布局和呈现。
您可以尝试重写DispatcherScheduler
的行为以自定义优先级进行计划,如下所示:
public class PriorityDispatcherScheduler : DispatcherScheduler
{
private readonly DispatcherPriority _priority;
public PriorityDispatcherScheduler(DispatcherPriority priority)
: this(priority, Dispatcher.CurrentDispatcher) {}
public PriorityDispatcherScheduler(DispatcherPriority priority, Dispatcher dispatcher)
: base(dispatcher)
{
_priority = priority;
}
public override IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
{
if (action == null)
throw new ArgumentNullException("action");
var d = new SingleAssignmentDisposable();
this.Dispatcher.BeginInvoke(
_priority,
(Action)(() =>
{
if (d.IsDisposed)
return;
d.Disposable = action(this, state);
}));
return d;
}
}
然后通过将ObserveOnDispatcher()
替换为ObserveOn(new PriorityDispatcherScheduler(p))
来修改可观察序列,其中p
是适当的优先级(例如,Background
)。
此外,这看起来非常可疑:ToObservable(NewThreadScheduler.Default)
. 我相信这将导致每次结果出现时创建一个新线程,其唯一目的是将其传递给调度程序,之后新线程将终止。 这几乎肯定不是你的意图。 我假设您只是想在单独的线程上处理文件;正如所写的那样,如果您的IEnumerable
产生 1,000 个项目,您的代码实际上最终会创建 1,000 个短期线程,而这些项目实际上都不会执行读取文件的工作。
最后,是否在调度程序线程上调用OpenFile()
? 如果是这样,我相信将要发生的事情如下:
- 调度程序(在UI线程上)将调用
Subscribe()
,它将处理可观察运算符的链一直回到ParseDataObservable(file)
。 - 调度程序将循环访问您的
IEnumerable
序列,将每个结果触发到ToObservable()
创建的可观察序列中。
传递 - 到可观察序列的每个结果都将被安排在调度程序(当前正在运行的同一调度程序)上传递。
如果是这种情况,那么在任何结果传递给AddIhvToList()
之前,将读取整个文件,因为调度程序被束缚在读取文件上,并且在完成之前不会处理其队列中的结果。 如果发生这种情况,您可以尝试更改代码,如下所示:
var obs = _parser.ParseDataObservable(file)
.ToObservable()
.SubscribeOn(/*NewThread*/Scheduler.Default)
.ObserveOnDispatcher() // consider using PriorityDispatcherScheduler
.Subscribe<Ihv>(AddIhvToList, ReportError, ReportComplete);
注入SubscribeOn()
应确保IEnumerable
的迭代(即文件的读取)发生在单独的线程上。 Scheduler.Default
在这里应该足够了,但如果你真的需要,你可以使用NewThreadScheduler
(你可能不需要)。 在设置完所有内容后,调度程序线程将从Subscribe()
返回,释放它以继续处理其队列,即在结果进入时将结果传递给AddIhvToList()
。 这应该为您提供所需的异步行为。