一个Observable()可以处理两种情况下的订阅吗？
Observable()是RxJS库中的一个关键类，用于创建可观察对象，它可以处理两种情况下的订阅。
- 单一值订阅：Observable()可以用于处理只有一个值的情况，例如一个API请求的响应结果。在这种情况下，Observable()会发出一个值并立即完成，订阅者可以通过订阅该Observable来获取该值。
- 多个值订阅：Observable()还可以用于处理多个值的情况，例如一个实时数据流。在这种情况下，Observable()会持续地发出新值，并且不会自动完成。订阅者可以通过订阅该Observable来获取连续的值，并可以选择在需要时手动取消订阅。
Observable()的优势在于它提供了一种简洁而强大的方式来处理异步数据流。它可以轻松地与其他RxJS操作符（如map、filter、reduce等）结合使用，以便对数据进行转换、过滤和聚合等操作。此外，Observable()还支持错误处理和取消订阅等功能，使得开发者能够更好地控制和管理数据流。
在腾讯云的产品中，与Observable()相关的产品是腾讯云的消息队列 CMQ（Cloud Message Queue）。CMQ提供了一种可靠的消息传递机制，可以用于在分布式系统中传递和处理异步消息。开发者可以使用CMQ来创建消息队列，并使用Observable()来订阅和处理队列中的消息。通过使用CMQ和Observable()的组合，开发者可以构建高效可靠的消息传递系统。
腾讯云CMQ产品介绍链接地址：https://cloud.tencent.com/product/cmq
相关·内容
我已经经历了很多线程，说需要取消订阅FormControl#valueChanges以防止内存泄漏。我理解了从Observable中取消订阅的“时间”和“如何”。据我所知，产生无限数量值的可观测值需要取消订阅，而FormControl#valueChanges就是这样一个Observable。
但我的问题是，为什么没有收集这些Observables的垃圾呢？我的意思是，当角分量被摧毁时，引用是死的，对吗？模板中的窗体控件不再存在。FormControl实例也消失了。那么，如果Observable的“源”FormControl消失了，那么为什么它的成员valueChanges仍然存在并保留其订阅
浏览 0提问于2018-08-15得票数 6
1回答
避免角内存泄漏的最佳实践
、、、、
我正在尝试检测和修复我的角/斯巴达克斯应用程序中潜在的内存泄漏。所以基本上我要找的是“订阅”调用我的代码库并评估我能做什么。
我知道，最简单的形式就是直接在html中使用可观察性，然后使用异步管道来读取它们，而不是调用订阅方法并将值分配给组件类属性。但有时候没那么简单。有时，可观察到的数据必须以某种方式准备好。
我的问题是这些方法是否同样有效？
将任何“订阅()”调用分配给订阅对象是一个好主意，我最终将在"ngOnDestroy"-method中取消订阅
为什么不能使用订阅对象调用代替“订阅”调用之后的“取消订阅”-method？
最重要的一点是:使用" pipe ()
浏览 2提问于2022-09-21得票数 2
我需要重新设置我的数据库中的数据，其中包括行动-
获取所有数据删除所有数据创建新的数据
按照顺序，我有以下三种方法-
getAllData(): Observable<IUser[]> {
//
}
clearAllData(users: IUser[]): Observable<boolean> {
//
}
generateNewData(cleared: boolean): Observable<IUser[]> {
//
}
在我的部件里，我用它们就像下面-
resetData(): void {
this.ser
浏览 4提问于2021-05-05得票数 0
回答已采纳
请考虑以下示例：
Observable.range(1, 10).subscribe(i -> {
System.out.println(i);
if (i == 5) {
throw new RuntimeException("oops!");
}
}, Throwable::printStackTrace);
这将输出从1到5的数字，然后打印异常。
我想要实现的是使观察者保持订阅，并在抛出异常后继续运行，即打印从1到10的所有数字。
我尝试过使用retry()和，但是，正如文档中所说的，它们的目的是处理可观察到的自身发出的错误
浏览 2提问于2014-08-16得票数 28
回答已采纳
考虑在Service中查询two服务的以下两种方法：
getAssessmentById(id: number): Observable<MotorAssessorReport> {
const url = environment.endPoints.assessment.base +
environment.endPoints.assessment.assessmentById + id;
return this.httpClient.get<MotorAssessorReport>(url)
}
async getAssessmentByIdAs
浏览 0提问于2019-04-28得票数 7
我在我的Angular应用程序中使用了Angular，我可以从DataService中观察到Angular组件，如下所示：
DataService.ts:
export class DataService {
private messageTracker = new BehaviorSubject<any>();
private fileTracker = new BehaviorSubject<any>();
getMessageTracker(): Observable<any> {
return this.m
浏览 0提问于2019-08-19得票数 0
回答已采纳
1回答
我有一个场景，当发生http端口错误时，效果被取消订阅，并且不再起作用。例如,
@Effect() newMessages$ : Observable<any> = this.actions$
.ofType(SEND_NEW_MESSAGE_ACTION)
.switchMap(action => this.threadsService.saveNewMessage(action.payload))
.catch(() => Observable.of(new ErrorOccurredAction("Error Ocurred while saving
m
浏览 0提问于2017-04-13得票数 2
下面是使用Func0和Action1的代码片段。
Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
try {
return Observable.just(Database.readValue());
}
catch(IOException e) {
return Observable.err
浏览 0提问于2018-03-21得票数 3
回答已采纳
1回答
我有一个关于Angular2+ Observable如何处理Http错误的问题。当我们使用Angular2+ Http进行请求时，如果返回的响应有错误状态码，就会抛出错误。
在Angular2官方启动器项目中，需要调用的错误处理函数：
someObs.catch(handler)
在处理程序中，它实际上返回：
Observable.throw(...)
我的问题是，这不会抛出新的错误，可能会导致web应用程序崩溃吗？因为在Javascript中，如果你抛出一些错误，web应用程序可能会出现意想不到的行为。
我对Observable.catch的理解是，当一些错误发生时，我们捕获它，然后做一些其
浏览 2提问于2017-07-12得票数 1
我试图熟悉反应性背压处理的问题，特别是通过阅读这个wiki：
，我们有一个更复杂的示例代码：
// we have to multicast the original bursty Observable so we can use it
// both as our source and as the source for our buffer closing selector:
Observable<Integer> burstyMulticast = bursty.publish().refCount();
// burstyDebounced will be our buff
浏览 3提问于2015-10-04得票数 5
回答已采纳
在使用可观察对象之后，是否需要释放/释放可观察对象，以防止可观察对象被GC垃圾化？我没有找到任何文章来提及这件事。可观测的物体会被GC自动地垃圾化吗？
Observable.create(new ObservableOnSubscribe<Boolean>() {
@Override
public void subscribe(ObservableEmitter<Boolean> e) throws Exception {
boolean connected = true;
//my lo
浏览 2提问于2017-06-30得票数 1
回答已采纳
1回答
在安卓系统上，我编写了一个Observable，它应该在2000毫秒之后被调用一次，但它从来没有被调用过。
Observable.timer(2000, TimeUnit.MILLISECONDS) // wait 2000 msec
.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<Long, Observable<?>>() {
@Override
浏览 2提问于2014-10-04得票数 2
回答已采纳
读取并使用以下CanActivate检查来查看用户是否允许导航到主页：
canActivate(): Observable<boolean> {
return this.authQuery.isLoggedIn$.pipe(
take(1),
switchMap((isLoggedIn) => {
if (!isLoggedIn) {
this.router.navigate(['/login']);
}
return of(isLoggedIn);
})
);
我想知道take(1)的目的是什么
浏览 0提问于2018-12-11得票数 0
回答已采纳
似乎他们返回相同的结果，为什么使用一个而另一个呢？
var computed = ko.computed(function(){return 'foo';});
var observable = ko.observable();
var observableArray = ko.observableArray();
console.log('computed isObservable', ko.isObservable(computed));
console.log('observable isObservable', ko.isObser
浏览 5提问于2015-04-28得票数 9
回答已采纳
我已经编写了一个函数，该函数返回一个可观察到的包装onSnapshot方法的函数。
function getObservable() {
return Observable.create(observer => {
firebase.firestore().collection('users').onSnapshot(
snapshot => observer.next(snapshot.docs),
);
});
})
}
我能够使用这个函数并获得如下更新的文档
const observable = getO
浏览 0提问于2018-10-17得票数 1
回答已采纳
2回答
考虑一下守则：
res: any;
getData(url: any) {
this.res = this.http.get(url);
}
ngOnInit(): void {
getData("url.com/file.json");
console.log(this.res)
}
在控制台里我得到
Observable {source: Observable, operator: ƒ}
如何获得json文件的内容而不是可观察到的内容？
浏览 7提问于2022-07-06得票数 0
回答已采纳
1回答
我正在学习.NET中的反应性扩展(rx)，我对“订阅”到底是什么以及什么时候使用它感到有些困难。
让我们从线程中获取一些样本数据：
using System;
using System.Reactive.Linq;
using System.Threading;
namespace ConsoleApp1
{
class Program
{
class Result
{
public bool Flag { get; set; }
public string Text { get; set; }
浏览 0提问于2018-08-22得票数 1
回答已采纳
我一直在编写代码，如下所示：
class WidgetService {
getWidgets() {
return this.authService.user.pipe( //authService.user is an Observable, that emits the currently authenticated users.
first(user => user!=null), //Make sure a null object isn't coming through
switchMap(user =&
浏览 0提问于2018-07-18得票数 2
回答已采纳
1回答
关于这部分代码，我遇到了一些问题：
SRNetwork.provider
.request(SRService.postData(user_id: userId))
.mapArray(STrain.self)
.observeOn(ConcurrentDispatchQueueScheduler.init(queue: SDispatchQueue.dataTrain.getQueue()))
.subscribe({ (event) -> Void in
switch event {
浏览 5提问于2017-01-02得票数 6
2回答
在角7中，我从API中获取和发布数据。
GET
为了获取数据，我使用了一个可观察的(posts$)和管道/地图；
我在组件HTML和中使用异步，不使用订阅；
这是我最近看到的方法，包括在角度文档中。
POST
Service Post方法还返回数据：
可能的验证错误或创建的Post的ID。
当API返回它们时，我需要获取其中的任何一个。
但是，当调用服务以发布以下内容时，没有任何作用：
this.postService.addPost(model).pipe();
除非我使用subscribe，如下所示：
this.postService.addPost(mod
浏览 2提问于2019-02-21得票数 4
function MyViewModel() {
var self = this;
this.firstName = ko.observable('Planet');
this.lastName = ko.observable('Earth');
this.computedState = ko.observable(false);
this.fullName = ko.computed({
read: function () {
console.log("READ"
浏览 1提问于2012-08-03得票数 1
回答已采纳
每次我从MainComponent导航到TestListComponent时，都会触发TestListComponent构造函数，并创建一个新的ObservableService实例。当我单击该链接时，控制台将显示重复的消息。可能是角度问题，有帮助吗？
main.module.ts
import { NgModule } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';
import {MainRoutingModule}
浏览 3提问于2017-02-27得票数 2
回答已采纳
1回答
我有一个问题，这是一个简单的业务逻辑流程：
检查是否有多个部门、部门和员工关系的员工，首先检查缓存中是否存在关系，如果存在，检查员工是否属于该员工，如果不在缓存中，则从数据库中获取，并检查该员工的关系，然后将部门信息保存到缓存中。
代码如下：
public Observable isEmployeeInDepartment(List<Long> departmentIds, long employeeId){
//this observable will resolve twice, and cause unnecessary cache access
Ob
浏览 2提问于2017-04-27得票数 0
我想在调用我的rx翻新服务之前检查手机是否连接到了互联网。如果没有连接，我想返回一个包含错误的假响应。
我以下面的解决方案结束，使用defer()，但我认为它可以更好，有什么提示吗？
private Observable<Response> checkNetwork(Observable<Response> retrofitService) {
return Observable.defer(new Func0<Observable<Response>>() {
@Override
浏览 0提问于2015-02-25得票数 9
我在我的安卓应用程序中使用RxJava，我想从数据库中加载数据。
通过这种方式，我使用Observable.create()创建了一个新的可观察到的EventLog列表。
public Observable<List<EventLog>> loadEventLogs() {
return Observable.create(new Observable.OnSubscribe<List<EventLog>>() {
@Override
public void call(Subscriber<? sup
浏览 2提问于2015-12-10得票数 32
回答已采纳
是否需要取消订阅Angular HttpClient的方法返回的Observable？
例如：
this.http.get(url).subscribe(x => this.doSomething());
我需要取消订阅此订阅吗？我问这个是因为我不知道Angular是否会自己处理它。此外，请求是一次性的，而不是持续的。我倾向于认为我不需要。你的想法是什么？
根据下面的帖子：
RxJS负责一次性订阅，但我在他们的文档中没有找到任何东西。
浏览 2提问于2019-10-18得票数 6
我不太明白下面的行为： let throw$ = Observable.throw(() => "Server error");
let testClick$ = Observable.fromEvent(this.saveButton.nativeElement, 'click')
.pipe(
switchMap(() => throw$),
finalize(() => console.log("finalize testClick$"))
);
let okClick$ = testClick$.pi
浏览 8提问于2019-06-20得票数 1
回答已采纳
我有时需要一次更新许多文档，10K甚至更多，我在想每种方式的优点/缺点是什么？
我想到了两种方法，我很乐意了解什么是优势/劣势，并听取第三种方法。
我正在寻找一个好的批量方式来做到这一点。
方法1，发送到observable，获取所有文档并完成工作：
Observable
.from(ids)
.flatMap(new Func1<String, Observable<JsonDocument>>() {
@Override
public Observable
浏览 1提问于2015-02-25得票数 1
我很难让Observer.onErrorResumeNext像我所期望的那样行事。
Subscriber orchestratorObserver = new Subscriber<Integer>() {
@Override
public void onCompleted() {
System.out.println("orchestratorObserver."+"onCompleted()");
}
@Override
public void onError(Throwable e) {
浏览 3提问于2016-11-29得票数 0
回答已采纳
我不理解Finally方法。它在这种情况下不会触发。
[TestMethod]
public void FinallyHappensOnError()
{
bool finallyActionHappened = false;
try
{
Observable
.Throw<Unit>(new DivideByZeroException())
.Finally(() => finallyActionHappened = true)
.Subscribe();
}
catch
浏览 0提问于2017-11-11得票数 2
我正在寻找最优雅的方式来完成以下任务。我知道有多种方法可以做到这一点，而且有多种方法遵循反模式(当然，这里我不想要:订阅中的订阅)。
我的圈套
export class MyService {
giveMeTheBest(...): Observable<TheBest> {
const observable1$ = // Fancy request, returning an Observable
const observable2$ = // A request to a different endpoint, also returni
浏览 1提问于2020-09-15得票数 0
回答已采纳
2回答
我对Rx非常陌生，只是想知道subscribeOn的顺序如何影响Observable
//This will not print anything
Observable.just("whatever")
.flatMap { s -> Observable.just(s.length) }
.subscribeOn(Schedulers.newThread())
.subscribe(::println)
//This prints the length
Observable.just("whatever"
浏览 4提问于2017-02-10得票数 1
回答已采纳
嗨，我正试着抓住角度9中的rxjs，并且我尝试根据一个可观察到的响应有条件地打一个电话，但是如果没有嵌套订阅，它似乎无法工作。我读过几篇文章，包括和，但我似乎无法解决我的问题。
在我的阅读中，我也读到了
当单击确认按钮时，将触发DialogConfirmService中的DialogConfirmService方法。
请注意，我已经尝试了两个以上的因素，但张贴的例子，我一直在尝试。
我当前的工作解决方案(我想摆脱嵌套订阅)
this.dialogConfirmService
.confirmed()
.pipe(take(1))
.pipe(map(x => {
浏览 3提问于2020-07-13得票数 1
回答已采纳
这两者之间的区别是(摘自)：
public static IObservable<TProperty> OnPropertyChanges<T, TProperty>(this T source, Expression<Func<T, TProperty>> property)
where T : INotifyPropertyChanged
{
return Observable.Create<TProperty>(o=>
{
var propertyName = property.G
浏览 2提问于2018-10-26得票数 2
回答已采纳
我正在学习Rx，我不明白为什么我的代码挂起了。这是简单的Rx观察，我想提供最后一个元素，当它完成。
var observable = Observable.Range(1, 20);
var results = new List<int>();
int lastElement = 0;
observable.Subscribe(results.Add, () =>
{
Console.Out.WriteLine("Completed " + results.Count);
lastElement = observable.Last(); //
浏览 1提问于2014-03-25得票数 0
2回答
我刚开始学习“角”，我所遵循的教程中有"Observable“这个术语。导师解释了，但我没有完全理解。
什么是Observable，为什么我们总是要调用observable.subscribe()
subscribe()实际上是做什么的？
浏览 1提问于2017-10-16得票数 2
回答已采纳
直接订阅从方法返回的Observable是一种好的做法吗？
在调用回调方法之前，它会被垃圾回收吗？
例如，我在服务中有一个方法：
get<City>(id): Observable<City> {
var url = this.baseUrl + "api/Cities/" + id;
return this.http.get<City>(url);
}
然后在组件中，我有：
ngOnInit () {
this.cityService.get<City>(this.id)
.sub
浏览 0提问于2020-04-13得票数 0
当我们已经取消订阅observable(具有setInterval).Even时，为什么我们需要用clearInterval返回函数?如果我不返回函数clearInterval，它会给下面的示例提供相同的result.Referring
我尝试删除返回线，结果相同
const evenNumbers = Observable.create(function(observer) {
let value = 0;
const interval = setInterval(() => {
if (value % 2 === 0) {
ob
浏览 2提问于2019-05-14得票数 2
1回答
我有一堂课：
export class Stuff {
id: string;
namd: string;
}
关于angular < 4.3，我有一个电话：
getStuff(): Observable<Stuff[]> {
return this.http.get('api-url-here/stuff')
.map((data: Response) => {
return data.json()
.map((item: any) =
浏览 4提问于2017-08-15得票数 0
回答已采纳
我正在使用一个可观察的间隔，每5秒在某个组件内执行某个函数，这与我的路由配置中的路径相关。
Observable.interval(5000).subscribe(res => {
// Something happens here
});
但是，当路径更改时，我希望终止间隔。现在它继续执行任务..。
做这件事最好的方法是什么？
浏览 3提问于2016-04-30得票数 7
回答已采纳
1回答
在我的体系结构中，我使用一个名为UseCase的类来组成一个可观察对象并订阅它。UseCase接受构造函数中的一个或多个存储库，并有一个"execute“方法，负责创建可观察对象并订阅它。
这里有一个简单的例子；
public class MyUseCase {
IRepoA mRepoA;
IRepoB mRepoB;
Scheduler mScheduleOn;
Scheduler mObserveOn;
CompositeSubscription mCompositeSubscription;
public MyUseCase(I
浏览 10提问于2017-03-11得票数 0
回答已采纳
1回答
我正在致力于一个角2应用程序，并需要一些指导如何处理身份验证错误干净。
我的最终目标是能够集中处理每个Http请求的身份验证错误(特别是401和403)。
我发现问题对我的开始非常有帮助，但是我仍然坚持用正确的方式为我的自定义Http实现返回的每个可观察到的错误处理程序注册。
下面是我目前正在处理的一个示例：
import { Injectable } from 'angular2/core';
import { Http, ConnectionBackend, Request, RequestOptions, RequestOptionsArgs, Response } fr
浏览 0提问于2016-01-28得票数 28
回答已采纳
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
回答已采纳
我正在创建一个在输入字段下显示验证错误的组件。如果显示了一条错误消息，并且用户提交了我想要闪现的消息，以引起他们的注意。
我想知道是否可以使用可观察对象作为输入绑定？
这样，我就可以订阅输入，并在观察到任何数据时进行闪存。
下面是我的想法的一个例子：
@Component({..})
export class MessageComponent implement OnChanges {
@Input()
public flash: Observable<any>;
public ngOnChanges(changes: SimpleChanges): void
浏览 4提问于2017-07-08得票数 5
回答已采纳
2回答
ngOnInit()中有Observable<>。 ngOnInit() {
this.products$ = this.someService.getMissingProducts();
} this.products$是Observable类型，并且产品具有代码(字符串类型)成员字段。我想要将所有Product[]的代码值更改为新的值。 对于这个愚蠢的问题我很抱歉。我不擅长typescript和angular，所以我试图找到一些解决方案，但失败了。 有什么简单的方法吗？ 提前谢谢。 附注:我将此代码添加到我的解决方案中。谢谢爱德华 this.products$
浏览 60提问于2021-01-09得票数 0
回答已采纳
2回答
我希望创建一个LocationHandler类，它返回一个observable<Location>，它可以发送一个新的位置，订阅者得到最后一个添加的位置和任何后续的值。
我写了这个类，但我不知道这是不是正确的方法，因为我添加了一个回调，我闻起来很难闻。
谢谢你的帮助。
public class LocationHandler {
private MessageHandler<Location> onNewItem;
private Observable<Location> locationObservable;
public Loc
浏览 4提问于2016-11-08得票数 0
回答已采纳
1回答
角度2两个观测值
、、、
我有一个组件，它调用服务来通过HTTP获取一些数据。不过，该服务需要在API中“登录”，如果它之前没有登录的话。
因此，当我从组件调用函数getFunnel()时，我希望该函数检查是否存在会话。如果没有会话，func应该'login‘来获取令牌，然后运行预期的http来返回正确的可观察值。
import { Injectable } from '@angular/core';
import { DatePipe } from '@angular/common';
import { Observable } from 'rxjs/Observabl
浏览 0提问于2018-01-27得票数 0
2回答
在角应用程序中，从服务器获取数据有不同的方法：
从服务中获得可观察性，并在组件上订阅它
在服务中创建主题，在组件上订阅主题
这两种方法都适用于我，但我不知道该用哪种方法。
First method.从服务中获得可观察性，并在组件上订阅它
article.service.ts
import { Injectable } from '@angular/core';
import { Article } from '../models/article';
import { AngularFirestore } from '@angular/fi
浏览 0提问于2019-08-24得票数 3
回答已采纳
2回答
可观测定时器处理
、、、、
我正在使用反应性.NET扩展，我想知道如何处理它。我知道在某些情况下，像这样处理它是很好的：.TakeUntil(Observable.Timer(TimeSpan.FromMinutes(x)))。我
第一例
在这种情况下，我有一个计时器，它在x秒之后触发，然后它就完成了，应该被释放。
public void ScheduleOrderCancellationIfNotFilled(string pair, long orderId, int waitSecondsBeforeCancel)
{
Observable.Timer(TimeSpan.FromSeconds(waitSe
浏览 5提问于2021-02-01得票数 1
回答已采纳
2回答
我有一个可观察的链，初始的可观察对象来自网络，并将在每次准备读取消息时触发。然后，下一个处理程序读取消息并反序列化它。现在我有了一个可观察对象的分支，一个是消息处理程序，另一个是记录消息。
问题是，因为我使用的是observable，所以我实际上会尝试读取消息两次。
我知道使用Event而不是Observable可以解决这个问题，但是我会遇到一个垃圾收集问题，可能会导致套接字不被收集。
我想到的一个解决方案是插入某种类型的分隔符，结束一条可观察对象链并创建一个新的链，这样的函数是否已经作为fsharp或其他库的一部分存在。
这个问题还有其他的解决方案吗？
编辑：
不能正常工作的代码示例
let
浏览 12提问于2017-08-25得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
