我正在尝试使用 RxJSrepeatWhen
运算符进行网络重试。这个想法是,当调度程序收到一个新请求时,我直接尝试该请求,如果它导致网络故障结果,我将它添加到一个池中,以便稍后重试。所以我的调度程序的入口点是队列函数,它完成这样的工作:
queue({ operation, body }) {
const behaviourSubject = new BehaviorSubject();
const task = {
operation,
body,
behaviourSubject,
};
this.doTask(task).subscribe({
error: _ => this.notifier.next({ tasks: [task], remove: false }),
complete: () => console.log('Task successfully done on first try: ', task),
});
return behaviourSubject;
}
并且this.notifier
是Subject
用作工人的通知者的。所以worker本身是这样的:
const rawWorker = new Observable(subscriber => {
const doneTasks = [];
const jobs = [];
for (const task of this.tasks) {
jobs.push(
this.doTask(task).pipe(
tap(_ => {
doneTasks.push(task);
}),
catchError(error => { task.behaviourSubject.next(error); return of(error); }),
)
);
}
if (jobs.length > 0) {
forkJoin(...jobs).subscribe(_ => {
this.notifier.next({ tasks: doneTasks, remove: true });
subscriber.next(`One cycle of worker done. #${doneTasks.length} task(s) done and #${this.tasks.length} remaining.`);
// subscriber.complete();
if (this.tasks.length > 0) {
this.notifier.next();
}
})
} else {
subscriber.complete();
}
});
const scheduledWorker = rawWorker.pipe( // TODO: delay should be added to retry and repeat routines
retry(),
repeatWhen(_ => this.notifier.pipe(
filter(_ => this.tasks.length > 0),
)),
);
并且通知器跟踪数组中的所有未完成的请求,如下所示:
this.notifierSubscription = this.notifier
.pipe(
filter(data => data && data.tasks)
)
.subscribe({
next: ({ tasks = [], remove = false }) => {
if (remove) {
console.log('removing tasks: ', tasks);
this.tasks = this.tasks.filter(task => !tasks.some(tsk => task === tsk));
} else {
console.log('inserting: ', tasks);
this.tasks.push.apply(
this.tasks,
tasks,
);
}
console.log('new tasks array: ', this.tasks);
}
});
据我所知,如果工人的一个周期没有完成,那么 repeatWhen 无关紧要。例如,如果我删除部分:
else {
subscriber.complete();
}
从工人那里,在工人第一次尝试(一个空周期)时,Observable 没有完成并且 repeatWhen 不会做任何事情。但另一方面,正如你所见,我已经评论过// subscriber.complete();
存在工作但重复发生的情况。问题的最糟糕的部分是许多不同的工作实例并行运行,这会产生许多重复的请求。
我在这个问题上花了很多时间,但没有任何线索可以追踪。