为什么将相同的 observable 传递给 concat() 多次以这种方式工作？

Question

我在文档上读到：

如果你多次传递 concat 相同的 Observable，它的值流将在每次订阅时“重播”，这意味着你可以重复给定的 Observable 任意多次。如果将同一个 Observable 传递给 concat 1000 次变得乏味，您总是可以使用 repeat。

https://rxjs.dev/api/index/function/concat

我的问题是：为什么它会这样工作？我希望它只返回第一个可观察的值，然后完成（因为所有其他的也会完成，因为我们正在引用到同一个可观察的）。以这种方式工作将与其他 concat() 行为一致。

Answer 1

考虑以下几点：

// Defining a function
const addTwoNumbers = a => b => () => {
  console.log(`Adding ${a} + ${b}`);
  return a + b;
}

// Defining another function
const increment = addTwoNumbers(1);

// Defining another function
const incFive = increment(5);

// Invoking the final function to get a result
const result = incFive();

// Result is 6
console.log(`Result is ${result}`);


// Invoke 3 more times
[incFive, incFive, incFive].forEach(
  fn => fn()
);

注意到每次调用incFive 时我们是如何运行代码1 + 5 的吗？如果你运行这段代码，你会看到

Adding 1 + 5
Result is 6
Adding 1 + 5
Adding 1 + 5
Adding 1 + 5

尽管函数只定义了一次，但函数中的代码总共“运行”了 4 次。

---

这就是 observables 的工作原理！ subscribe 说运行 observable，所以如果你订阅 10 次，你运行 observable 10 次。

当然，如果你正在处理一个“热”的 observable，这会有所改变。使用concat 重复一个监听按钮舔的可观察对象（例如）没有多大意义。然而，使用concat 从服务器请求更新更有意义。

定义一个 observable 一次并能够重用它是很好的。上面，“invoke 3 more times”下面的行可以重写为：

// Redefine 3 more times
[addTwoNumbers(1)(5),addTwoNumbers(1)(5), addTwoNumbers(1)(5)].forEach(
  fn => fn()
);

但这很麻烦！此外，有时您想在一个地方定义一个函数并在另一个地方完全调用它。到那时，您可能不知道如何重新定义它。可观察对象也是如此。

Answer 2

concat 的策略是订阅第一个提供的 observable，完成后 concat 订阅第二个 observable，以此类推。

第二个方面是您提供冷可观察对象。这意味着每次订阅完成时，observable 从头开始​​。更多关于冷热观测值的信息，你可以在这里找到What are the Hot and Cold observables?

这里是 marble diagrams 的测试游乐场，以便更好地理解。

const {concat} = rxjs;
const {take, mergeMap, tap} = rxjs.operators;
const {TestScheduler} = rxjs.testing;
const {expect} = chai;

const test = (testName, testFn) => {
  try {
    testFn();
    console.log(`Test PASS "${testName}"`);
  } catch (error) {
    console.error(`Test FAIL "${testName}"`, error.message);
  }
}

const createTestScheduler = () => new TestScheduler((actual, expected) => {
  expect(actual).deep.equal(expected);
});

test('should concat two cold observables', () => {
  const testScheduler = createTestScheduler();
  testScheduler.run((helpers) => {
    const { cold, hot, expectObservable } = helpers;
    const init$ = cold('       --e-------------------|');
    const letters$ = cold('      --a--b--c|           ');
    const numbers$ = cold('               --0--1--2|  ');
    const resultMarble = '     ----a--b--c--0--1--2--|';
    const result$ = init$.pipe(
      mergeMap(() => concat(letters$, numbers$))
    );
    expectObservable(result$).toBe(resultMarble);
  });
});

test('should conconcat two hot observables', () => {
  const testScheduler = createTestScheduler();
  testScheduler.run((helpers) => {
    const { cold, hot, expectObservable } = helpers;
    // the ^ character marks the zero time of hot observable
    const init$ = cold('       ----e---------------|');
    const letters$ = hot('-----^--a--b--c--d--------');
    const numbers$ = hot('---0-^----1--2--3--4------');
    const resultMarble = '     ------b--c-3--4-----|';
    const result$ = init$.pipe(
      mergeMap(() => concat(
        letters$.pipe(take(2)),
        numbers$.pipe(take(2))
      ))
    );
    expectObservable(result$).toBe(resultMarble);
  });
});

<script src="https://cdnjs.cloudflare.com/ajax/libs/chai/4.1.2/chai.js"></script>
<script src="https://unpkg.com/rxjs@^7/dist/bundles/rxjs.umd.min.js"></script>