RXJS 6 如何实现每500ms重复获取图表水平线并输出至Observable流

• 对delay操作符的作用理解偏差：delay(500)只会把Observable已经推送出去的所有值，顺延500ms再发给下游，完全不会驱动你的取数方法重复执行。按照你现在的写法，订阅之后会立刻同步跑完所有取形状、遍历的逻辑，把所有价格通过next()发出，之后等500ms，这些暂存的值才会真正到达下游。
• 对repeat的触发条件认知错误：repeat只有在收到上游发送的complete()完成通知时，才会重新订阅上游、重复执行一遍上游逻辑。你自定义的Observable构造函数里，跑完所有next()之后根本没有调用observer.complete()，上游永远不会发送完成通知，repeat完全不会触发，自然实现不了每500ms重复取数的效果。
• 定时逻辑的放置位置错误：RxJS6中要实现“固定周期重复执行一段逻辑”，不能靠delay，需要从流的最起始位置放置一个定时触发源，比如timer或interval。

• 用timer(0, 500)作为整个流的触发起点：第一个参数0代表订阅后立刻执行一次取数逻辑，不用等第一个500ms周期过去才拿到初始的形状列表；第二个参数500代表首次执行后，每隔500ms再触发一次取数，完全匹配你的取数节奏要求。如果用interval(500)，首次触发要等500ms，会出现明显的初始延迟。
• 用switchMap把定时触发的信号映射到实际取数逻辑上：每次定时器触发节拍时，调用取数方法拿到当前的水平直线价格列表。如果你的取数逻辑是同步的，switchMap完全够用；如果取数是异步的、担心上一次取数还没跑完下一次就触发导致重入，换成exhaustMap即可。
• 加一层distinctUntilChanged做优化：如果两次轮询拿到的水平直线价格列表和上一次完全一致，就不重复向下游推送值，避免每500ms就重复输出一遍没有变化的内容，刚好匹配你提到的场景要求——没有新增直线时不重复输出，新增价格20的直线后，流里才会追加对应的新值输出。

import { timer } from 'rxjs';
import { switchMap, distinctUntilChanged } from 'rxjs/operators';

// 封装当前获取所有水平直线价格的逻辑，返回全量价格数组
const getCurrentHorizontalLinePrices = () => {
  const prices: number[] = [];
  chart.activeChart().getAllShapes().forEach(shape => {
    if (shape.name === 'horizontal_line') {
      chart.activeChart().getShapeById(shape.id).getPoints().forEach(point => {
        prices.push(+point.price.toFixed(4));
      });
    }
  });
  return prices;
};

// 组装Observable流
const test$ = timer(0, 500).pipe(
  // 每次定时触发时执行取数
  switchMap(() => getCurrentHorizontalLinePrices()),
  // 价格列表无变化时跳过输出
  distinctUntilChanged((prevPrices, currPrices) => JSON.stringify(prevPrices) === JSON.stringify(currPrices)),
  // 将数组中的每个价格逐个对外输出
  switchMap(prices => prices)
);

• 初始状态下有3条价格为5、10、17的水平直线：订阅后立刻执行第一次取数，拿到价格数组[5,10,17]，和初始空值对比有变化，依次输出5、10、17，符合预期。
• 后续每次500ms轮询时，如果形状没有变化，拿到的价格数组和上一次完全一致，会被distinctUntilChanged过滤，不会重复输出。
• 当新增一条价格为20的水平直线后，下一次轮询拿到的数组变为[5,10,17,20]，和上一次的数组有差异，会依次把全量价格输出，流中自然包含新增的20，和你提到的场景完全匹配。

内容的提问来源于stack exchange，提问作者MrYutz