我有一个 angualr 7 应用程序，总共有 20 个传感器数据。我想使用 observable 为选定的传感器 ID 每 5 秒接收一次数据。例如；
var sensorId = ""; // dynamically selected from the web UI
var sensorData$ = interval(5000).pipe()
sensorData$.subscribe() // etc..
我将选择多个传感器，并开始使用间隔订阅获取数据。我怎样才能保留这些可观察的？我该如何管理它？
另外，我可以随时添加任何传感器。
【问题讨论】：
标签： javascript angular typescript rxjs angular7
我建议使用 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);
}
这合适吗？我会根据你的需要更新我的代码，在评论区告诉我吧。
【讨论】：
你可以使用 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 完成时才发出值，它只发出一次值。
再次记住管理您的退订。
【讨论】：