如何在 Angular 7 中管理多个 observable

Posted 2023-03-13

【中文标题】如何在 Angular 7 中管理多个 observable

【英文标题】：How to manage multiple observable in angular 7

【发布时间】：2019-11-26 04:11:47

【问题描述】：

我有一个 angualr 7 应用程序，总共有 20 个传感器数据。我想使用 observable 为选定的传感器 ID 每 5 秒接收一次数据。例如；

var sensorId = ""; // dynamically selected from the web UI
var sensorData$ = interval(5000).pipe()
sensorData$.subscribe() // etc..

我将选择多个传感器，并开始使用间隔订阅获取数据。我怎样才能保留这些可观察的？我该如何管理它？

另外，我可以随时添加任何传感器。

【参考方案1】：

我建议使用 Subject 和 mergeMap。

至于解释，您将拥有 1 个主题，当用户从 UI 中选择一个主题时，您将在其中发出新的 sensorId。然后，您需要订阅该确切的主题，并且在 mergeMap 的帮助下，您将拥有订阅方法中的所有传感器值。

让我们看看演示代码：

private sensorIdSubject = new Subject();

ngOnInit() 
  this.sensorIdSubject.pipe(
    mergeMap(sensorId => 
      interval(5000).pipe(switchMap(() => this.getSensorData(sensorId)))
  ).subscribe(sensorData => 
    // Every 5 second, each of the sensorIds that have been selected in UI will 
    // get sensor data and emit new value here in subscribe, because all of them
    // have been merged.
  )


public chooseSensor(sensorId) 
  this.sensorIdSubject.next(sensorId);

这合适吗？我会根据你的需要更新我的代码，在评论区告诉我吧。

【讨论】：

首先，非常感谢您的好评！我只需要从订阅方法中获取相关数据，然后我必须更新相关的图表系列。但是，我想知道很多订阅者攻击会更新同一张图表会发生什么。这个操作是线程安全的吗？

另外，我可以直接订阅 interval() 吗？在这种状态下，我可以在不知道接收到哪些数据而不是合并的订阅方法的情况下更新图表 int 相关的订阅方法。

还有一个要求，我可以随时删除任何传感器。因此，我必须在需要时取消订阅相关传感器。

【参考方案2】：

你可以使用 rxjs forkJoin

它将你所有的 observable 组合成一个发出所有值的，所以你订阅了 forkJoin 发出的 observable，你可以添加管道操作符等等

类似：

let sensors = forkJoin(observable1, observable2)

sensors.subscribe(
   sensorData => 
      // You receive an object with your observables data
   
)

如果您需要获取可变数量的可发出不同值的可观察对象，您可以这样使用 BehaviorSubject：

let sensors = new BehaviorSubject<Array<Observables<yourType>>([])

sensors.next([newArrayOfObservables])

sensors.subscribe(
    sensors$ => 
      sensor$.forEach(sensor => 
        sensor.subscribe( // Do your logic)
      
    
)

这样你可以添加/删除 observables，因为 ForkJoin 只在 observables 完成时才发出值，它只发出一次值。

再次记住管理您的退订。

【讨论】：

那么，我可以随时向 forkJoin 添加一个新的 observable 吗？例如，forkJoin 接受一个可观察的列表。我已经开始 forkjoin 间隔，然后我将新的 observable 添加到 observable 列表中。在这种状态下，forkjoin 中是否会收到新的 observable 的数据？

我认为在这种情况下你可以做的是创建一个 BehaviourSubject，然后在每次更新你的 observable 列表时发出一个 forkJoin observable，这样你将始终订阅最后一个 observable 列表，请记住管理您的退订以避免内存泄漏

但是这些 observables 与其他的不同。因此，一个可观察的结果不必等待其他可观察的结果

非常感谢。我接受我的要求的另一个答案。