RxJS检测长时间运行的订阅
RxJS是一个响应式编程库，用于处理异步数据流和事件流。它提供了丰富的操作符和工具，使开发者能够更轻松地处理复杂的异步场景。
RxJS的核心概念是Observable（可观察对象），它代表了一个异步数据流或事件流。开发者可以对Observable进行各种操作，如过滤、转换、合并等，以实现对数据流的处理和转换。
对于长时间运行的订阅，RxJS提供了一些机制来检测和处理。其中一个重要的机制是使用Subscription（订阅）对象来管理订阅的生命周期。通过调用Subscription对象的unsubscribe()方法，可以取消订阅并释放相关资源，避免内存泄漏。
另外，RxJS还提供了一些操作符和工具，用于处理长时间运行的订阅。例如，可以使用timeout操作符设置一个超时时间，如果订阅超过指定的时间仍未完成，可以触发相应的处理逻辑。还可以使用retry操作符在订阅失败时进行重试，以确保订阅能够成功。
RxJS在前端开发中有广泛的应用场景，特别适用于处理复杂的异步操作，如网络请求、用户输入事件、定时器等。它可以帮助开发者简化异步代码的编写和管理，提高代码的可读性和可维护性。
腾讯云提供了云函数SCF（Serverless Cloud Function）服务，可以与RxJS结合使用，实现更灵活和高效的云计算应用。通过使用SCF，开发者可以将RxJS的异步处理能力与腾讯云的弹性计算资源相结合，实现高可用、高性能的云计算应用。
更多关于腾讯云云函数SCF的信息和产品介绍，可以访问以下链接：
总结：RxJS是一个用于处理异步数据流和事件流的响应式编程库。对于长时间运行的订阅，可以使用Subscription对象管理订阅的生命周期，并通过timeout和retry等操作符处理超时和重试。腾讯云的云函数SCF可以与RxJS结合使用，实现高效的云计算应用。
相关·内容
我们的角应用程序中有许多组件，它们需要每秒钟定期显示每个组件所特有的新值(计数、时间戳、运行时间等)。最自然的方法是创建使用RxJS、timer和interval工厂函数的可观察性。然而，这些触发角度变化检测在整个应用程序的每一个间隔上的次数与调用间隔函数的次数一样多。如果页面上有几十个组件，那么每秒钟或每隔一段时间就会触发对整个应用程序的更改检测数十次，从而造成很大的性能开销。
到目前为止，我有两种方法试图解决这个问题。一个好的答案对两者都会很有帮助--最好两者都是。我希望避免手动触发更改检测，而是依赖可观察到的新值，并让异步管道/OnPush更改检测策略负责触发更改检测。如果这是不可能的，
浏览 1提问于2018-12-04得票数 9
回答已采纳
一段时间以来，我一直在想角的问题，上周我决定深入研究。它有相当惊人的功能和库，但有点复杂的架构。我现在正在研究ngrx图书馆，以便完全理解它。但在那之前，
有件事让我很困惑却搞不懂为什么？
在我的组件中，我有一个按钮来加载诸如Api服务分配给我的任务这样的项目。我获取数据并将其记录到控制台，但直到我第二次单击按钮时模板才会更新它。我错过了什么？
谢谢。
// Angular
import { Component, OnInit, ElementRef, ViewChild, ChangeDetectionStrategy, OnDestroy } from '@angular/cor
浏览 0提问于2019-04-14得票数 0
回答已采纳
我正在使用Angular 5和SignalR构建一个应用程序。我创建了一个管理集线器的服务，每当有事件从服务器传入时，它都会将值放入BehaviorSubject私有变量中。有一个只读属性，它应该输出我可以从组件订阅的可观察对象。
在组件中，我已经订阅了服务的Observable，但它仍然不会在属性更新时更新。
我知道"test“属性正在更新，因为我必须告诉我，我从服务器收到了配置。
有没有人能指出我可能遗漏了什么，或者不完全理解Angular句柄是如何改变检测的，或者我是如何破坏它的？
服务：
import { Injectable, Input, Output, EventEmit
浏览 1提问于2018-01-30得票数 5
我在组件树的某个地方创建了一个500 ms的Observable.interval，并订阅了它。该组件没有输入或输出属性。该间隔触发每次发送滴答时从根组件开始的更改检测。这在我的应用程序中造成了很多不必要的开销。我没有找到任何关于这种行为的文件。
是否有可能关闭这种可观察到的变化检测？
编辑：添加代码
下面的代码演示了我想要做的事情。我按照Günter的建议在角区域外放置了间隔，但是现在对数组的修改没有在模板中发布。有没有任何方法可以在不触发更改检测的情况下更新模板？
import {NotificationList} from "./NotificationList";
im
浏览 0提问于2016-09-02得票数 6
2回答
我的预加载器代码是： import { Injectable } from "@angular/core";
import { Subject, Observable } from 'rxjs';
@Injectable()
export class SpinnerService {
private preloaderState$ = new Subject<boolean>();
activate() {
this.preloaderState$.next(true);
}
disactivate() {
t
浏览 27提问于2020-04-24得票数 0
在rxjs中，如果调用了complete方法，我是否也应该调用取消订阅？
例如，我有一个来自rxjs的timer，我将其设置为5秒。
在调用订阅函数后，我是否也应该从计时器运行取消订阅？
import { timer } from "rxjs";
const token = timer(5 * 1000).subscribe({
next: () => {
console.log("xxx");
token.unsubscribe(); // <---- should I do it to to free memory?
浏览 2提问于2021-03-23得票数 0
2回答
我对角和RxJs和Ngrx都很陌生
我知道，我们赞同一个可观察的，并观察变化。我在Ngrx 中看到了这段代码
<div>Current Count: {{ count$ | async }}</div>
问题是，到底是什么?它和订阅一个可观察到的有什么不同？一种什么时候用，什么时候用？
浏览 1提问于2019-05-06得票数 3
回答已采纳
1回答
考虑以下类型记录/ rxjs代码：
import { range } from 'rxjs';
range(1, 5).subscribe(async (num) => {
await longLastingOperation(num);
})
function longLastingOperation(num) {
console.log('starting', num);
return new Promise((resolve) => {
setTimeout(() => {
console.log(
浏览 0提问于2018-11-07得票数 0
回答已采纳
4回答
我有重复的角度问题，但我搜索了很多关于这个问题，并使用了所有的技术，在堆叠溢出和.。
我的问题是当我订阅一个以上的加载程序组件时。
这是我的装载机组件
import { Component, ChangeDetectionStrategy, ChangeDetectorRef, DoCheck, OnChanges, AfterViewInit, OnInit } from '@angular/core';
import { Subject, BehaviorSubject } from 'rxjs';
import { LoaderService } from
浏览 0提问于2019-09-01得票数 0
回答已采纳
我在我的rxjs应用程序中注意到，rxjs操作符，如switchMap、map等，不起作用。(我的IDE也给出了一个提示。)当然，我运行了以下代码。如果我将switchMap替换为subscribe，就会在控制台中看到“返回的值”。
知道运营商为什么不工作吗？
import { Component, OnInit } from '@angular/core';
import {Observable} from "rxjs/Observable";
import 'rxjs/add/operator/switchMap';
@Component(
浏览 1提问于2016-12-05得票数 1
回答已采纳
1回答
我对rxjs有点陌生(我们已经使用angularJS很长时间了，终于摆脱了我们的技术债务)。
使用angular7，我有一个这样的请求列表。使用forkJoin，我这样管理它：
let requests = []
for (let asset in this.assets)
{
if(asset.enabled)
requests.push(this.apiService.postAsset(oAsset))
}
return forkJoin(requests)
而且效果很好，我得到了我的一系列反应。然而，，，我有点想弄清楚如何根据资产类型获得不同的回调。我想要这样的
浏览 0提问于2019-03-12得票数 0
回答已采纳
1回答
我有以下代码来管理用户，
@Component({
selector: 'app-user',
template: `
<input [(ngModel)]="newUser"><button (click)="addNewUser()">add</button>
<div *ngFor="let user of users$ |async">
{{user}}
</div>
`
})
export class UserComponent
浏览 0提问于2019-02-06得票数 1
1回答
我有问题的主题和评论系统，我正在建立。它运行得很好，但是直到有人发布另一条评论时，第一条评论才会出现，在此之后，所有其他评论都可以正常工作。我试过使用BehaviourSubject给出的初始值为null或"“和ReplaySubject，但我想我可能遗漏了一些关于它们的工作原理，有没有办法让这项工作实时进行，即当用户推送评论时，它会到达服务器并被添加回堆栈，而不需要立即提交第二条评论？
import { Injectable } from '@angular/core';
import { Observable } from "rxjs/Observable&
浏览 1提问于2019-08-13得票数 1
我使用的是具有响应形式的角5，需要使用valueChanges来禁用所需的动态验证。
组件类：
export class UserEditor implements OnInit {
public userForm: FormGroup;
userName: FormControl;
firstName: FormControl;
lastName: FormControl;
email: FormControl;
loginTypeId: FormControl;
password: FormControl;
confirmP
浏览 6提问于2017-12-14得票数 45
回答已采纳
2回答
尝试用一些数据向我的服务器发布一个简单的帖子，并让它返回一些JSON数据。(目前，它在呼叫中运行，但我在我的网络选项卡上没有看到它真的试图进行吗？)
我的服务(api.service.ts)
import { Injectable } from '@angular/core';
import { Headers, Http, Response} from '@angular/http';
import {Observable} from 'rxjs/Rx';
// Import RxJs required methods
import
浏览 1提问于2016-12-28得票数 12
回答已采纳
所以，基本上，rxjs是新的，想知道，
我是否需要从Observable.of(数据)或更新的rxjs版本中取消订阅(数据)？
即
import {of} from 'rxjs';
const sub = of(data).subscribe();
这有必要吗？
sub.unsubscribe();
浏览 2提问于2019-04-28得票数 6
1回答
我使用的是角4+，并且我注意到在可观察到的解析立即(同步地？)调用Observable.subscribe()之前，角没有更新绑定的值。在其他地方，这很好，因为网络调用将可观察到的完成延迟足够长时间，使初始绑定变化以角形式生效。
//our root app component
import {Component, VERSION} from '@angular/core';
import {Observable} from 'rxjs/Observable';
import 'rxjs/add/observable/of';
import &
浏览 1提问于2018-04-18得票数 2
回答已采纳
我有一个消息服务，每当调用API方法时都会发出一条消息。我的想法是，我的应用程序中的所有其他组件都可以调用服务中的方法来显示错误或成功消息。
import { Injectable } from '@angular/core';
import { MessagingComponent } from ',/messaging.component';
import { ReplaySubject } from 'rxjs';
@Injectable()
export class MessageService {
public messag
浏览 0提问于2016-12-28得票数 1
使用Angular，我是RXJS观察点的新手，并试图了解这个错误。我有以下基本代码： export class EventService {
test$ = new Subject();
watch() {
return this.test$;
}
push() {
this.test$.next('Hello World');
}
}
export class Component {
watch() {
this.eventService.watch().subscribe((obj)=>{ console.log(&#
浏览 68提问于2019-10-13得票数 1
回答已采纳
2回答
我从组件中调用一个服务，该服务返回json数据，下面是我的代码
import {Component,Input,Output} from 'angular2/core';
import {PostService} from './post.service';
import {OnInit} from 'angular2/core';
import {Observable} from 'rxjs/Observable';
@Component({
selector: 'postC',
template: `
<
浏览 4提问于2016-11-23得票数 0
回答已采纳
2回答
角9从可观测的角度返回
、、、、
我正在寻找一种从另一个可观察到的返回一个可观察的方法。我发现您可以使用pipe和map操作符来实现这一点，但它似乎不适合我。我做错什么了？
我使用角9.1.12和rxjs ~6.5.4。
示例: Service1
import { Observable, of } from 'rxjs';
export class Service1 {
test(): Observable<string> {
console.log(1);
return of('hey');
}
}
Service2
import { Obser
浏览 1提问于2021-07-15得票数 0
回答已采纳
今天，我用RxJS在Node.js中编写了一些代码，得到了非常好的结果:程序运行try{}块，但根本没有运行finally{}块，在try{}块中使用代码0退出。
在使代码简短之后，为了重现此问题，步骤如下：
首先创建一个名为BehaviorSubject subject的
调用subject.asObservable().share()并获得一个可观察的名为obs的
调用obs.subscribe()
调用await obs.first().toPromise()
最后，该程序表现出一种非常有线的行为:退出try {}块而不运行下面的finally {}块中的代码。
浏览 0提问于2018-04-09得票数 2
回答已采纳
我有以下服务来适应我的应用程序中的全局旋转器：
import {
Injectable
} from '@angular/core';
import {
Observable
} from 'rxjs/Observable';
import {
ReplaySubject
} from 'rxjs/ReplaySubject';
@Injectable()
export class SpinnerService {
private visible = new ReplaySubject < boolean > (
浏览 6提问于2017-05-16得票数 1
回答已采纳
1回答
用户功能从未调用
、、
我试图在服务器上查询一个字符串令牌，然后客户端组件将使用该令牌。客户端上的meteor方法调用期望一个异步结果，并返回一个可观察的结果。一旦该方法完成，它将调用可观察到的next。组件订阅可观察到的内容。
但是，当代码运行时，存在一个问题。在调试器中，观察者没有任何订阅者。因此，在调用next()时，不会调用订阅者回调。
我不清楚是怎么回事，我会感激另一副有经验的眼睛。
import { Component, OnInit, OnDestroy, Injectable } from '@angular/core';
import { Subject } from 'rx
浏览 3提问于2017-05-02得票数 1
1回答
在RxJS4中，以下代码将冻结您的环境：
Rx.Observable.of(10)
.repeat()
.take(1)
.subscribe(value => {
console.log(value);
});
这是因为它将在即时调度程序上运行repeat，并且永远不会到达take。现在在RxJS5中没有立即调度程序，但是如果我运行相同的代码，一切都会按预期工作。究竟是什么改变了，我们在运行什么调度程序？我认为默认情况下默认不是调度程序，如果我正确理解的话，这也会冻结repeat。
浏览 4提问于2017-03-16得票数 2
1回答
我们有一个无法完成的rxjs订阅。我想知道这是rxjs库中的错误，还是我们做错了什么。
我们的代码如下所示：
private loadGroups(): void {
if (this.groupOptionsSubscription?.closed === false){
this.groupOptionsSubscription.unsubscribe();
}
this.groupOptionsSubscription =
this.lookupService
.getGroupLookups(null, this.fi
浏览 1提问于2021-11-15得票数 1
我想在一个函数中运行代码--然后作为一个websocket Observable返回。有效监控长时间运行的进程。我不知道如何通过websockets以这种格式正确返回值。 我的长时间运行的流程：(显然不会花很长时间) import { Observable } from 'rxjs';
export function longRunningProcess (): Observable<unknown> {
return new Observable(subscriber => {
subscriber.next('End
浏览 30提问于2020-11-09得票数 0
回答已采纳
下面是components.ts中一个角的代码，下面是用X编译的错误代码。
错误
合并/merge.Component.ts:75:12-错误TS2551:属性‘un订阅’在类型‘可观察’上不存在。你是说“订阅”吗？
$().unsubscribe()~
../../node_modules/rxjs/dist/types/internal/Observable.d.ts:53:5 53订阅(观察者:部分)：订阅；~‘订阅’在这里声明。
import { Component, ElementRef, OnDestroy, OnInit, ViewChild } from '@angul
浏览 8提问于2021-12-17得票数 1
回答已采纳
1回答
从另一个类调用init函数，则从不调用console.log。稍后，调用broadcast会产生以下错误：
TypeError:无法读取未定义的属性“next”
带有可观察代码的文件：
import 'rxjs/add/operator/filter';
import 'rxjs/add/operator/mergeMap';
import 'rxjs/add/operator/share';
import 'rxjs/add/observable/of';
import { Observable } from '
浏览 1提问于2017-09-12得票数 2
3回答
如何完成可观察性？
、、、、
学习rxjs玩它。我的代码：
// RxJS v6+
import { withLatestFrom, map } from 'rxjs/operators';
import { interval } from 'rxjs';
const source = interval(1000);
const example = source.pipe(
map(value => value +1),
map(value => {
if(value === 40) {
finish();
}
else if (v
浏览 1提问于2019-12-02得票数 5
回答已采纳
我正在用angular2开发应用程序，这里我使用消息广播将数据从一个组件更新到另一个组件。
这是代码，message_event.ts
import {Injectable} from '@angular/core';
import {Observable} from 'rxjs/Observable';
import {Broadcaster} from './broadcaster';
@Injectable()
export class MessageEvent {
constructor(private broadcaster: B
浏览 4提问于2017-09-05得票数 1
1回答
我目前正在学习rxjs，发现示例操作符并不像我所希望的那样工作。这是我的示例代码。
const source = rxjs.interval(500);
rxjs.interval(300)
.pipe(
rxjs.operators.sample(source)
)
.pipe(
rxjs.operators.take(20)
)
.subscribe(val => console.log(val));
<script src="https://cdnjs.cloudflare.com/ajax
浏览 5提问于2020-12-06得票数 0
回答已采纳
使用rxjs，我有一个昂贵的计算，它映射出BehaviorSubject的结果。
const obs$ = sameBehaviorSubject.pipe(map(expensiveComputation));
然后，在我的代码的其他部分中，我订阅了可观察到的内容。每次我订阅，它都会重新运行expensiveComputation。我怎么才能防止这种情况？
浏览 0提问于2018-04-26得票数 4
回答已采纳
2回答
在我的项目中有许多使用异步/等待的代码。但是我想把所有的代码都转到rxjs代码上。
但我不习惯用rxjs。请告诉我将异步/等待代码转换为rxjs的模式或方法。或者请推荐一个例子。
这是我的原始代码(使用异步/等待)
this.userHistory$ = new Subject();
async getUserStatue(userId, query){
const userData = await this.UserApi.getUserInfo(userId, query);
this.userHistory$.next(userData);
}
下面的代码是我试图使用rxjs
浏览 7提问于2022-05-16得票数 0
我想实现一个搜索组件，如角教程。
问题:
search方法触发并修改searchTerms: Subject<string>，但switchMap从不触发。
代码：
进口
import { Component, OnInit } from '@angular/core';
import { SearchService } from "./search.service";
import { Observable } from 'rxjs/Observable';
import { Subject }
浏览 2提问于2017-07-26得票数 0
回答已采纳
1回答
我在catchError中使用了rxjs，并且熟悉pipe流中使用的pipe操作符，特别是对于HttpClient (XHR)调用
我的问题是catchError操作是如何工作的？它是如何捕捉引擎盖下的错误的？
import { throwError, of } from 'rxjs';
import { catchError } from 'rxjs/operators';
//emit error
const source = throwError('This is an error!');
//gracefully handle erro
浏览 0提问于2018-12-01得票数 2
回答已采纳
1回答
如演示所示，如标题所示
const { combineLatest, interval, of } = rxjs;
const { first, last, sample, take, withLatestFrom } = rxjs.operators;
const numbers = interval(1000);
const takeFourNumbers = numbers.pipe(take(4));
takeFourNumbers.subscribe(x => console.log('Next: ', x));
setTimeout(()=>{
浏览 4提问于2022-03-12得票数 0
如果观察体完成，我是否仍然必须取消订阅/处置(在Rxjs中)观察体以移除观察者(防止内存泄漏)，或者一旦发生onComplete或onError事件，这是否由Rxjs内部处理？
浏览 0提问于2018-10-01得票数 2
1回答
我试图在一个可以观察到的简单代码中抛出一个错误：
return Observable.create((observer) => {
observer.throw(new Error('Test'));
});
然而，出于某种原因，throw还没有定义。我不知道为什么在我们班的第一名，我有以下的重要内容：
import 'rxjs/observable/throw';
我也尝试过：
import 'rxjs/Observable/throw';
以及：
import 'rxjs/add/observable/throw';
浏览 5提问于2016-09-01得票数 0
回答已采纳
我有一个创建倒计时的Observable.timer函数，当计时器在我的@组件中结束而不使用setTimeout()时，我想调用一个名为endTimer()的特定函数。我可以在视图中检查计数器== 0的值，但是如何签入@组件
import { Component } from '@angular/core';
import { IonicPage, NavController, NavParams, ModalController } from 'ionic-angular';
import { Storage } from '@ionic/storag
浏览 1提问于2018-03-26得票数 0
1回答
我正试着打http电话，如果会有任何错误，做我的事情。
这是我的代码：
import { Injectable } from '@angular/core';
import { Http, Response } from '@angular/http';
import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/throw';
// Operators
import 'rxjs/add/operator/catch';
浏览 0提问于2016-12-05得票数 0
回答已采纳
1回答
考虑一个简单的情况，您需要过滤一个数组。在角度和rxjs的上下文中，我们可以使用JS filter方法和RxJS filter运算符。
//JS filter
let array = ['1', '2', '3', '4']
let result = array.filter(x => x == '2')[0];
//RXJS filter
const p = from(array);
let result: any;
p.pipe(filter(x => x == '2')).sub
浏览 1提问于2018-08-28得票数 0
我对角和RxJS相当陌生，我很难了解RxJS操作符的某些方面。我这里有一些代码(在组件中)：
this.salesService.getOrders().subscribe(orders => {
this.orders = orders;
this.orders.forEach((order, index) => {
this.salesService.getOrderCustomers(order.id).subscribe(customers => {
this.orders[index]['customerName'] =
浏览 11提问于2022-10-13得票数 1
回答已采纳
2回答
我有一个应用程序需要发送小的rxjs请求来重新加载其他组件中的某些方法(这些组件在导航树中的位置要高得多，这样输入和输出就太混乱了)。我有一个子组件，它向服务发送一个布尔值：
子组件：
myMethod(){
this.reloadComponentService.changeReload(true)
};
重发服务：
import { Injectable } from '@angular/core';
import {BehaviorSubject} from 'rxjs';
@Injectable()
export class ReloadCompon
浏览 1提问于2022-03-07得票数 1
回答已采纳
在app.component中，我需要基于ActivatedRoute执行一个操作。使用: Angular6，rxjs6 (用它的新语法)
import {Observable} from 'rxjs';
import {map} from 'rxjs/operators';
....
constructor (private route:ActivatedRoute){
const url: Observable<string> =route.url.pipe(map(segments=>segments.join(''
浏览 0提问于2018-05-17得票数 0
回答已采纳
我在角8中创建父/子关系。Parrent组件调用设置可观察roleList的服务调用，然后将此roleList传递给子组件。在子组件中，我尝试订阅子组件接收的输入变量，并使用新值实例化MatDataTableSource。
问题是：子组件接收值，但组件不会重新呈现。在调用应用程序中的更改(例如:将鼠标悬停到另一个更改状态的组件)之后，它确实显示了从服务接收到的数据。
这些是构成部分：
父组件.ts:
import { Component, OnInit, ChangeDetectionStrategy } from
'@angular/core';
import { Role
浏览 3提问于2019-09-30得票数 2
回答已采纳
2回答
我创造了一个从5到0的倒计时。当您单击“开始”按钮时，它就会启动：
<html>
<head>
<meta charset="utf-8"/>
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.js">
</script>
</head>
<body>
<button id="start">START</button>
COUNT
浏览 1提问于2020-05-26得票数 1
回答已采纳
1回答
从我的部分开始，我宣布.
import { Observable } from 'rxjs/Observable';
import { Subscription } from 'rxjs/Subscription';
import { tap, map, concat } from 'rxjs/operators';
contactsDataSource$: Observable<any>;
然后，我将流分配给来自http服务的可观察到的流：
this.contactsDataSource$ = this.medialistsrv.
浏览 1提问于2018-05-08得票数 4
回答已采纳
import { of, Subject, interval } from 'rxjs';
import {
tap,
startWith,
map,
delay,
publishReplay,
publish,
refCount,
withLatestFrom,
switchMap,
take
} from 'rxjs/operators';
const a$ = new Subject();
const b$ = a$.pipe(
map(a => {
console.log('expen
浏览 45提问于2021-07-06得票数 4
1回答
我刚刚了解了RxJS对象的内存泄漏。因此，我创建了一个小代码来测试当我取消订阅主题时是否正常。下面是我的代码：
let subj = new rxjs.Subject();
const sub1 = subj.subscribe(text => console.log(text));
subj.next('subject 1');
subj.unsubscribe();
subj = new rxjs.Subject();
const sub2 = subj.subscribe(text => console.log(text));
subj.next('s
浏览 4提问于2019-12-29得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
