1

我正在尝试使用 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();存在工作但重复发生的情况。问题的最糟糕的部分是许多不同的工作实例并行运行，这会产生许多重复的请求。

我在这个问题上花了很多时间，但没有任何线索可以追踪。

4

0 回答 0