我正在尝试使用 Rx 并行处理项目。看来我无法告诉 Rx 并行运行我的观察者的 OnNext() 。
这是测试代码来演示
[Test]
public void ObservableObserveOnNewThreadRunsInParallel()
{
Console.WriteLine("Starting thread: {0}", Thread.CurrentThread.ManagedThreadId);
// store items as they are output
var list = new List<Tuple<string, int, int, int, TimeSpan>>();
// used to wait until a sequences are complete
var ev = new AutoResetEvent(false);
// try these schedulers
var schedulers = new[] {
Tuple.Create("ThreadPoolScheduler.Instance", (IScheduler)ThreadPoolScheduler.Instance),
Tuple.Create("NewThreadScheduler.Default", (IScheduler)NewThreadScheduler.Default),
Tuple.Create("TaskPoolScheduler.Default", (IScheduler)TaskPoolScheduler.Default),
Tuple.Create("Scheduler.Default", (IScheduler)Scheduler.Default),
Tuple.Create("Scheduler.Immediate", (IScheduler)Scheduler.Immediate),
};
// try each scheduler
foreach (var schedulerTuple in
schedulers) {
// emit tuples <i, delay> where delay decreases with each new tuple
// such that output timing is expected to be reversed
var observable =
Observable.Range(0, 5)
.Select(i => Tuple.Create((int)i, (int)(500 - i * 100)))
.Take(5);
var dt = DateTime.Now;
Tuple<string, IScheduler> scheduler = schedulerTuple;
observable
// specify the scheduler to use
.ObserveOn(schedulerTuple.Item2)
.Subscribe(
v => {
// emulate some work (first items take longer than last items)
Thread.Sleep(v.Item2);
// record when the item is done recording
lock (list)
list.Add(
Tuple.Create(
scheduler.Item1,
v.Item1,
v.Item2,
Thread.CurrentThread.ManagedThreadId,
dt - DateTime.Now));
},
// let the test go on
() => ev.Set());
// wait until the end of the sequence
ev.WaitOne();
}
// print observed order
foreach (var i in list) {
Console.WriteLine(i);
}
}
和输出:
Starting thread: 5
(ThreadPoolScheduler.Instance, 0, 500, 9, -00:00:04.2514251)
(ThreadPoolScheduler.Instance, 1, 400, 9, -00:00:04.6524652)
(ThreadPoolScheduler.Instance, 2, 300, 9, -00:00:04.9524952)
(ThreadPoolScheduler.Instance, 3, 200, 9, -00:00:05.1525152)
(ThreadPoolScheduler.Instance, 4, 100, 9, -00:00:05.2525252)
(NewThreadScheduler.Default, 0, 500, 11, -00:00:06.5066506)
(NewThreadScheduler.Default, 1, 400, 11, -00:00:06.9066906)
(NewThreadScheduler.Default, 2, 300, 11, -00:00:07.2067206)
(NewThreadScheduler.Default, 3, 200, 11, -00:00:07.4067406)
(NewThreadScheduler.Default, 4, 100, 11, -00:00:07.5067506)
(TaskPoolScheduler.Default, 0, 500, 12, -00:00:00.5020502)
(TaskPoolScheduler.Default, 1, 400, 12, -00:00:00.9020902)
(TaskPoolScheduler.Default, 2, 300, 12, -00:00:01.2021202)
(TaskPoolScheduler.Default, 3, 200, 12, -00:00:01.4021402)
(TaskPoolScheduler.Default, 4, 100, 12, -00:00:01.5021502)
(Scheduler.Default, 0, 500, 13, -00:00:00.5020502)
(Scheduler.Default, 1, 400, 13, -00:00:00.9020902)
(Scheduler.Default, 2, 300, 13, -00:00:01.2021202)
(Scheduler.Default, 3, 200, 13, -00:00:01.4021402)
(Scheduler.Default, 4, 100, 13, -00:00:01.5021502)
(Scheduler.Immediate, 0, 500, 5, -00:00:00.5020502)
(Scheduler.Immediate, 1, 400, 5, -00:00:00.9040904)
(Scheduler.Immediate, 2, 300, 5, -00:00:01.2041204)
(Scheduler.Immediate, 3, 200, 5, -00:00:01.4041404)
(Scheduler.Immediate, 4, 100, 5, -00:00:01.5041504)
请注意,即使我明确使用,每个 OnNext 调用似乎都在等待上一个调用ObserveOn()
指定用于通知的调度程序。
我希望除 Scheduler.Immediate 之外的所有通知都并行运行。
有人知道我做错了什么吗?