我认为你需要这样的东西：

subscribe(): void {
  this.word.debounceTime(400)
  .distinctUntilChanged()
  .switchMap((words: string): Observable<any[]> => 
    Observable.forkJoin(words.split(' ').map(word => this.http.get(word))) 
  )
  .map(arrayWithArrays => [].concat(arrayWithArrays)
  .subscribe((result: any[]): any[] => this.result = result);
 }

我想出了这个巨大但部分的解决方案：

在这里小提琴：https : //jsfiddle.net/mtawrhLs/1/

Rx.Observable.prototype.combineAllCont = function () {
    let lastValues = [];

    return Rx.Observable.create(obs => {
        let i = 0;

        let subscription = this.subscribe(stream => {
            const streamIndex = i;
            subscription.add(stream.subscribe(res => {
                lastValues[streamIndex] = res;
                obs.next(lastValues);
            }));
            i++;
        });

        return subscription;
    });
}

/** STUFF TO DEMO **/
let searchBox = [
    '',
    'car',
    'car ',
    'car food',
    'car food sun',
    'cat food sun',
    'cat sun'
]

function performSearch(word) {
    return Rx.Observable.defer(() => {
        console.log('sending search for ' + word);
        return Rx.Observable.of({
            word: word,
            results: word.split('')
        });
    })
}

let searchBoxStream = Rx.Observable.interval(500)
    .take(searchBox.length * 2)
    .map((i) => searchBox[i % searchBox.length]);

/** GOOD STUFF STARTS HERE **/
let wordsStream = searchBoxStream
    .map((str) => str.trim().split(' ').filter((w) => w.trim() != ''));

let wordSearchSubjects = [];
let wordSearchStreamSubject = new Rx.ReplaySubject(1);

wordsStream.subscribe((words) => {
    const nWords = words.length;
    const nSubjects = wordSearchSubjects.length;
    // Update streams
    for (i = 0; i < nWords && i < nSubjects; i++) {
        wordSearchSubjects[i].next(words[i]);
    }

    // Create streams
    for (i = nSubjects; i < nWords; i++) {
        const wordSearchSubject = new Rx.ReplaySubject(1);
        wordSearchSubjects.push(wordSearchSubject);
        wordSearchStreamSubject.next(
            wordSearchSubject.asObservable()
                .distinctUntilChanged()
                .flatMap((w) => performSearch(w))
                .concat(Rx.Observable.of(false)) // Ending signal
        )
        wordSearchSubjects[i].next(words[i]);
    }

    // Delete streams
    for (i = nWords; i < nSubjects; i++) {
        wordSearchSubjects[i].complete();
    }
    wordSearchSubjects.length = nWords;
});

let wordSearchStream = wordSearchStreamSubject
    .combineAllCont()
    .map((arr) => arr.filter((r) => r !== false));

resultingStream = wordSearchStream
    .map((arr) => {
        let ret = [];
        arr.forEach(search => {
            ret.push(search.results);
        });
        return ret;
    })
    .subscribe((arr) => console.log(arr));

需要改进的地方：

1. 我不得不使用combineAll不等待原始流完成的自定义实现。如果没有正确实现，这可能会导致内存泄漏，并且它应该删除完成的内部订阅（现在没有）
2. 必须改进词变化检测：从中间删除一个词会使一半的词再次被搜索。

可能有一个更好、更直接的解决方案，但我只能想出这个烂摊子。感觉非常hacky/危险。

我认为解决方案非常明确。首先，我们必须对单词执行差异算法。需要删除陈旧的词，只会请求新词。

唯一真正的问题是我们如何在没有任何数据竞争的情况下以最干净的方式实现它。所以我想出了一个实现：

searchTags$.pipe(
  scan((pair, tags) => [pair[1] || pair[0], tags], [[]]),
  concatMap(([prevTags, nextTags]) =>
    from(createActions(prevTags, nextTags))
  ),
  reduce((collection$, action) => collection$.pipe(action), of([]))),
  switchMap(stream => stream)
).subscribe(collection => { /*...*/ })

function createActions(prevTags, nextTags) {
  nextTags.reduce((actions, tag) => {
    if (isOld(prevTags, tag)) {
      actions.push(removeAction(tag));
    }
    if (isNew(prevTags, tag)) {
      actions.push(addAction(tag));
    }
    return actions;
  }, []);
}

function removeAction(tag) {
  return map(collection => { collection[tag] = undefined; return collection; });
}

function addAction(tag) {
  return switchMap(collection => requestTag(tag).pipe(map(
    res => { collection[tag] = res; return collection; }
  )));
}

这是我能做的最短的！