订阅作为Observable提供的Observable -> Filter ->
Map -> Subscribe的流程是什么？
订阅作为Observable提供的Observable -> Filter -> Map -> Subscribe的流程是一个典型的响应式编程流程，用于处理数据流的转换和订阅。
- 订阅：首先，我们需要订阅一个Observable对象，Observable是一个可观察的数据源，可以产生一系列的事件或数据流。
- Filter：在订阅之后，我们可以对Observable产生的数据流进行过滤操作。Filter操作符可以根据指定的条件过滤掉不符合要求的数据，只保留满足条件的数据。
- Map：在过滤之后，我们可以对数据流进行映射操作。Map操作符可以将数据流中的每个数据项进行转换，生成一个新的数据流。
- Subscribe：最后，我们需要订阅经过过滤和映射处理后的数据流。通过Subscribe操作符，我们可以定义对数据流的处理逻辑，例如打印数据、保存数据或进行其他业务逻辑的处理。
这个流程可以帮助我们对数据流进行处理和转换，使得我们能够更方便地处理和利用数据。在实际应用中，这个流程可以用于各种场景，例如数据过滤、数据转换、事件处理等。
腾讯云相关产品推荐：
- 腾讯云云函数（Serverless）：https://cloud.tencent.com/product/scf
- 腾讯云消息队列 CMQ：https://cloud.tencent.com/product/cmq
- 腾讯云云数据库 MySQL 版：https://cloud.tencent.com/product/cdb_mysql
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm
- 腾讯云人工智能平台 AI Lab：https://cloud.tencent.com/product/ai_lab
相关·内容
我使用的是角度9，我试图在一个可观察到的地方过滤值，但是它没有影响。我认为我处理可观察性的方式是错误的。
下面，_filter函数应该过滤this.filteredApproverOptions，但是我发现没有应用过滤器，并且始终填充整个列表。
private setFilteredApproverOptions(): void {
for (const field in this.approvalEditFormGroup.controls) {
this.approvalEditFormGroup.get(field).valueChanges.subscribe(
浏览 2提问于2020-07-17得票数 0
我有一个Observable<List<Event>>，我想让多个订阅者共享这个可观察对象。每个订阅者将过滤每个事件并对其进行处理。 Observable<List<Event>>是这样创建的： @Override
public List<Event> findNewEvents() {
List<Event> results = new ArrayList<>();
while(! fetchedEvents.isEmpty()) {
浏览 67提问于2020-11-25得票数 0
我在TypeScript v2.7.1中使用了角5，编译时我的client-details.component.ts抛出了这个错误：
src/app/components/client-details/client-details.component.ts(43,21)：ERROR TS2552中的错误:无法找到名称“events”。你是说“事件”吗？
上下文：ngOnInit从clientService调用getClient(id)，它应该返回我们订阅的类型：Event[] (DocumentsfromFireStore)。
客户端-细节.组件.：
import { Componen
浏览 0提问于2018-03-24得票数 0
回答已采纳
1回答
到目前为止，我对Rxjs的理解是，Observable是一个流，我们可以使用Observer的next方法在流中“推送”新值。然后我订阅该流来接收这些值。但我无法理解管道是如何工作的。例如。
我可以创建一个Observable
var observable = Rx.Observable.of('foo', 'bar');
那我就能做到
var subscription = observable.subscribe(x=>console.log(x))
在何处以及如何在上面创建观察者？
是谁在调用next以在流中推送值？
我认为x=>c
浏览 0提问于2018-05-22得票数 2
回答已采纳
1回答
在我的应用程序运行之前，我必须做一些配置，例如安装一个异常处理程序。因为我的应用程序有多个入口点(一个Activity，一个SyncAdapter)，所以我只想在我的Application类中做一次这个配置，并在入口点(Activity，SyncAdapter)上观察它。如果在入口点开始时配置已经完成，那么有关结果的信息仍然是可用的。
这是Observable类中的Application (RxJava 2)：
ConnectableObservable<Boolean> observable = Observable.create(…)
.subscribeOn(
浏览 3提问于2016-10-27得票数 0
const subscription: Subscription;
const first$ // Observable A
const second$ // Observable B
// When `first$` emits, I am checking the value and
// subscribe to `second$` if value is more than 5
subscription.add(
first$.subscribe(value => {
if(value > 5) {
subscri
浏览 21提问于2019-12-02得票数 0
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
如何获得基于id的内容？
当id为1(例如localhost:5555/product/1 )时，如何获得内容？我的路由文件已经安装好了
我有一个json文件，如
[
{
"id": "0",
"name": "xxxx",
"icon": "https://wikidownload.com/Download/xxxx.png",
"Website":"http.xxx.com"
},
{
"id":
浏览 0提问于2017-07-18得票数 0
2回答
敲除js计算不被解雇
、、、
代码可以在上找到。
在FilterViewModel中，我正在创建一个可观察的对象。
var FilterViewModel= ko.observable({
Name: ko.observable("test"),
Code: ko.observable("test"),
Number: ko.observable("test")
});
然后，在BankViewModel中，我运行一个计算方法，当任何输入框更改时，它都会触发。
var BankViewModel = function(){
var self
浏览 1提问于2014-04-30得票数 2
回答已采纳
这使用的是RxJava版本0.19.6。
在groupBy操作之外，可以创建由以下代码描述的管道，例如，根据某些条件从可观察的记录中选择一个记录，或者选择满足某些替代条件的第一个记录：
Observable<Long> observable = Observable.interval(1, TimeUnit.MILLISECONDS).take(10);
Observable<Long> filter1 = observable.filter(new Func1<Long, Boolean>() {
@Override
public Bool
浏览 0提问于2014-08-19得票数 2
目前还不清楚如何将错误传播到REactiveX中的订阅者，从而使可观察对象不会被销毁。
示例
observable.onNext(1);
observable.onNext(2);
observable.onError("Nope");
observable.onNext(3);<<won't work.
我接受这个限制，但是我仍然有这样的场景，我希望下游的侦听器知道发生了错误，并且我不希望可观察对象死亡。
这样做的主要用例是UI代码，如果出现错误，我不想对以前注册的所有可观察对象调用"Setup“。
可能的替代方案是
a)推送包含数据字段和错误字段
浏览 1提问于2016-06-23得票数 7
下面的代码运行良好。
但是，我认为combineLatest in other observable ($fromArray)的运算符(map)是很奇怪的。
我如何使combineLatest成为observable本身？
const item = ["television", "pc", "radio", "camera"];
const fooMarket = webSocket("wss://api.fooMarket.com/websocket/");
const barMarket = webSocket
浏览 12提问于2022-07-24得票数 0
回答已采纳
1回答
嗨，我有一个我不想公开的私有Subject，所以我正在创建与单个Subject不同的Observable。
public numberChange$: Observable<number>;
public oddChange$: Observable<number>;
public evenChange$: Observable<number>;
private number: Subject<number>;
constructor() {
this.number = new Subject<number>();
this.
浏览 0提问于2018-05-23得票数 0
回答已采纳
2回答
这个rxjs合并逻辑可以简化吗？
、、、、
我从Firebase客户端获得了一个可观察的(onAuthStateChanged)，即：
如果用户没有登录，会立即发出null，而会在几分钟后发出null，然后在用户登录时发出user object。
const observable = new Observable((obs) => {
return app.auth().onAuthStateChanged(
obs.next,
obs.error,
obs.complete
)
})
我想要的是：
accepted)always 忽略应用生命周期前1000 is发出
浏览 3提问于2021-08-09得票数 3
回答已采纳
使用RxJS和角2处理服务器重连接的最佳方法是什么？
到目前为止，这就是我所拥有的：
import { Injectable } from '@angular/core';
import { Http, Response } from '@angular/http';
import { Observable } from 'rxjs/Observable';
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import 'rxjs/add/operator/
浏览 4提问于2017-08-04得票数 1
回答已采纳
2回答
我正在尝试筛选一个列表，该列表可能是空的，也可能不是空的(或者该项不在列表中)。inboxData被另一个可观察到的填充：
private BehaviorSubject<InboxResponse> inboxData = BehaviorSubject.create();
public Observable<Item> getInboxItem(String id) {
return inboxData
.flatMap(response -> Observable.from(response.getData()))
浏览 6提问于2017-09-08得票数 5
回答已采纳
1回答
我正在开发一个简单的搜索屏幕来学习角订阅，我发现这是非常令人困惑的。在我的主页上创建了两个组件。一个过滤器组件和第二个搜索结果组件。
这是我的过滤服务
export class FilterService {
private filtersubject = new Subject<Filter>();
public currentfilters: Observable<Filter>;
constructor() {
this.filtersubject = new BehaviorSubject<Filter>(F
浏览 2提问于2022-01-19得票数 1
回答已采纳
我试着用CanActivate护卫的角度--当我从它和return true甚至return Observable.of(true);调用服务时，它不起作用
但是，当我不调用任何服务，只需用Observable.of(true);编写canActivate时，它就可以工作并加载组件。
以下是代码所做的工作：
canActivate(route: ActivatedRouteSnapshot, state: RouterStateSnapshot): Observable<boolean> {
//return Observable.of(true);
return
浏览 4提问于2017-09-19得票数 1
回答已采纳
有没有一种方法可以迭代javascript对象上的属性并测试可观察到的属性，这样我就可以在它上创建订阅了吗？下面只是一个简单的例子，但我的对象有大约30个属性。
var name = {
firstName: ko.observable(),
lastName: ko.observable()
};
// Iterate properties and test for observable to create subscription on.
lastName.subscribe(function(s) {
// modified
);
firstName.subscr
浏览 1提问于2018-12-12得票数 0
回答已采纳
1回答
我有一个函数，当它在其他地方被订阅时，我需要返回一个可观察对象，但因为我的函数异步地发出请求，然后试图将该数据作为可观察对象返回，所以它正在中断。 下面是它当前外观的一个示例： function userObservable() {
const select = 'select * from users';
this.db.query(select).subscribe(users => {
// apply some logic and filter to users here
// function just returns
浏览 18提问于2021-11-10得票数 0
回答已采纳
我在rxjs上很新，所以请耐心点:)。
var source = Rx.Observable.fromEvent(document, 'keyup');
source.filter(function(x){
console.log('filter with', x);
return true;
});
var subscription = source.subscribe(
function (x) {
console.log('Next: keyup!',x.keyCo
浏览 5提问于2016-07-26得票数 1
回答已采纳
1回答
我有一个包含键/值对的map。从这个map中，我将它们存储到一个作为结果Observable<String>返回的数据库中。
当我试图在不阻塞的情况下访问此字符串并将其放在列表中时，我的问题就出现了。
它始终是空的。下面是我的实现：
public Observable<List<String>> uploadKeys(Map<String, String> entries) {
final List<String> collection = new ArrayList<>();
entries.
浏览 2提问于2017-12-20得票数 2
回答已采纳
我已经创建了一个简单的方法，它将发送消息(作为对象)给给定的主题。在另一端，订阅服务器将使用该对象。
我将对多个用户使用相同的主题，因此在发送消息时希望发送一些附加信息，即200,300。这样消费者就可以过滤掉不相关的信息。但不知怎么搞不出来该怎么做。
附加信息(int参数)可以在postToSubject方法和action1处理，但不确定如何将其放在这里subject.onNext(object)
如果你能想出任何解决办法，请提出建议。
public void postToSubject(Object object, Subject subject) {
Observable<
浏览 5提问于2017-01-15得票数 3
回答已采纳
尝试为BufferUntil实现Observable (即，每当满足某些条件时，缓冲区元素和发出缓冲列表)
为RxNET实现
界定：
public static IObservable<IList<TSource>> BufferUntil<TSource>(
this IObservable<TSource> source,
Func<TSource, bool> predicate)
{
var published = source.Publish(
浏览 5提问于2017-02-16得票数 2
回答已采纳
我有一个愚蠢的问题:我正在实现一些包含其他可观察订阅的路由警卫，它们是在根中提供的，所以它们应该是单例的，不需要真正取消订阅。但是，我所做的是一直返回一个新的可观察到的新值，即我手动完成(不确定是否需要)，内部订阅将使用私有_sub字段清除每个执行。
问题是:这段代码是否安全，不受内存泄漏的影响(路由保护是如何在内部订阅的)？有一种推荐的方法来测试和看看在一个应用程序中是否有可观察到的未订阅的？
@Injectable({
providedIn: 'root'
})
export class MyRedirectGuard implements CanActivate
浏览 11提问于2022-08-22得票数 0
回答已采纳
我正在处理一个文件加密和上传类使用角。这些操作中有许多是异步的，因此我编写的方法正在返回RxJS可观测值。
// 1.
private prepareUpload(file): Observable<T>;
// 2.
private encryptData(data, filekey): Observable<T>
// 3.
private uploadEncryptedData(formData, token, range): Observable<T>
// 4.
private completeUpload(updatedFilekey, t
浏览 1提问于2019-03-29得票数 12
回答已采纳
我看到的是：
const observable = new Observable(subscriber => {
subscriber.next(1);
subscriber.next(2);
subscriber.next(3);
})
我想以这样的方式订阅：
observable.subscribe({
next(x) {
console.log('got value ' + x);
if (x == 2) {
const observable2 = new Observable(subscriber => {
浏览 1提问于2019-10-04得票数 0
2回答
例如，我的应用程序中有一个简单的Manager，我试图保存所有的reactive
class AppLockManager {
private val logger = LoggerFactory.getLogger(javaClass)
private val rxHelper: RxHelper
private val securityManager: DiarySecurityManager
private var locked = false
private var lastUnlockTime: LocalDateTime? = null
浏览 3提问于2016-09-19得票数 3
1回答
我正在使用可观察到的两个流和选择器函数的combineLatest。
const todos = Rx.Observable.of([{
id: 1,
completed: false,
text: 'one'
}, {
id: 1,
completed: true,
text: 'two'
}]);
const filter = Rx.Observable.of('SHOW_ALL');
const getTodos = Rx.Observable.combineLatest(todos, filter, (t
浏览 3提问于2017-06-02得票数 1
回答已采纳
我有一个外部可观测的，我使用它的结果在内部可观察，我需要从内部可观察的结果返回。
在下面的示例中，我需要返回第二个可观察到的结果allPersons
该函数的结果是可观察的，我希望它能够返回可观测的结果。
getAllPerson(): Observable<Data1[]> {
return this.dataService.getIds().subscribe(
(ids) => {
return this.dataService.getPersons().pipe(
map((allPersons) => {
浏览 4提问于2022-08-09得票数 0
我对这里一无所知。我正在尝试使用安卓的RxJava库，但是遇到了错误。逻辑是处理按钮上的点击事件，然后将其映射为从EditText获取值，然后将其flatmap到Single<String>，后者通过网络解析值。我尝试使用Schedulers.io()在IO线程上运行flatmap，但遇到错误： E/MainActivity: Login error
java.lang.IllegalStateException: Expected to be called on the main thread but was RxCachedThreadScheduler-1
浏览 60提问于2019-09-21得票数 1
回答已采纳
1回答
当前，当将值复制到领域时，我执行以下操作：
public void addToRealm(Home item, RealmChangeListener<E> listener) {
realm.executeTransaction((Realm realm1) ->
realm1.copyToRealm(item).addChangeListener<Home>(listener));
}
然后，我可以访问侦听器中新添加的对象。实现相同目标的合适的RxJava方法是什么？可观察到的对象必须返回Observable<Home&
浏览 0提问于2016-08-04得票数 2
回答已采纳
2回答
我只想在html中还原一个valure：
<tr *ngFor="let valCorr of valueCall| async">
<th>{{valCorr }}</th>
</tr>
在ts中
valueCall:Observable<number[]>
//this.studentService.getResult() return an Observable<number[]>
this.valueCall:Observable=this.studen
浏览 5提问于2022-10-19得票数 0
1回答
我希望删除登录的当前用户，这就是我使用筛选器的原因。但我得到了一个错误，就是：
类型“订阅”丢失了“Observable[]>”类型中的以下属性：_isScalar、源、操作符、lift等等。
下面是代码：
@Select(UserPageState.get('collection')) users$: Observable<Array<Partial<User>>>;
async ngOnInit() {
const USER = this.creds.credentials['id'];
this.user
浏览 2提问于2020-03-16得票数 0
当我从这个页面导航到另一个页面时，我需要停止Observable.timer。下面是我在我的项目中使用的代码。
import { Component, ChangeDetectorRef, OnDestroy } from '@angular/core';
import { NavController, MenuController } from 'ionic-angular';
import {Observable, Subject}from 'RxJs/Rx';
export class SettingsPage extends BaseP
浏览 4提问于2017-03-06得票数 0
我有一个简单的代码，返回缓存的公司或从数据库的公司。问题是我几乎在同一时间从两个不同的地方订阅了这个观察点。这会触发对数据库的2次调用，所以我只想在第一次调用完成并且缓存值已经存在的情况下运行第二次订阅。有没有通过rxjs运算符执行此操作的标准方法？
public current(): Observable<Company> {
if (this.cachedCompany) {
return Observable.of(this.cachedCompany);
} else {
if (this.
浏览 0提问于2018-05-25得票数 1
1回答
如何在Angular中返回在injectable上调用的订阅？ 所以我一开始花了一个小时才弄明白的是： returnURLNotes(): Observable<any> {
return this.store.select(selectDentureDesignNotes).subscribe((notes) => {
if (
notes &&
notes[this._CANVAS_NAME] &&
notes[this._CANVAS_NAME].length >
浏览 12提问于2019-04-21得票数 1
回答已采纳
我有一个从后端获取数据的类DataService：
@Injectable({
providedIn: 'root'
})
export class DataService {
constructor(private http: HttpClient) {}
getEntries() : Observable<Array<DataEntry>> {
return this.http.get<Array<DataEntry>>('http://192.168.0.103:3000/data')
浏览 6提问于2021-09-07得票数 0
应用程序实现了以下组件
customer-overview.component.ts
import { Component, OnInit, ElementRef, ViewChild } from '@angular/core'
import { Router } from '@angular/router'
import { DataSource } from '@angular/cdk'
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import { Obs
浏览 4提问于2017-08-18得票数 3
回答已采纳
2回答
我正在尝试创建一个具有起始值的订阅中心。我无法设置起始值。
现在我有了observable，它每隔30分钟生成一次DateTime.UtcNow。
public async Task Subscribe()
{
var client = Clients.Client(Context.ConnectionId);
var subscription = new CompositeDisposable();
var dateTimeUTCNow = Observable.Interval(TimeSpan.FromMinutes(30))
.Subscr
浏览 11提问于2019-07-16得票数 1
回答已采纳
我有多个可观察到的值，它们在页面的生存期内发出值。例如：
chartData$: Observable;
tableData$: Observable;
filterData$: Observable;
在任何时候，用户都可以单击“下载”按钮，并获得JSON，并组合上一次从这些可观察到的值中发出的值：
downloadButtonClicked$.pipe(
combine chartData$, tableData$ and filterData$ // <- how do I get latest values here?
).subscribe(([chart, ta
浏览 2提问于2020-09-08得票数 2
回答已采纳
我正在使用Observable.zip并行处理几个可观察到的数据。
Observable.zip(observable1, observable2).subscribe(([res1, res2]) => doStuff())
这里observable1和observable2是http请求，我有一个拦截器捕获错误。
import { Injectable } from '@angular/core';
import {
HttpInterceptor,
HttpHandler,
HttpRequest,
HttpErrorResponse
浏览 0提问于2018-06-29得票数 1
回答已采纳
1回答
我认为，由于我的误解，这是反应性编程的一个概念性问题。
我有一个Observable，它发出一个网络调用返回JSON，我filter()查看JSON的管理标志，然后在subscribe()方法中执行操作。
observable
.filter((json) -> { return json.isAdmin; })
.subscribe((json) -> { /* Do things for admin */ });
我也想这样做，但是当用户不是管理员时：
observable
.filter((json) -> { return !json.isAdmin; })
.sub
浏览 1提问于2016-03-03得票数 3
回答已采纳
2回答
我有一个没有完成的源流，但是我需要一个操作符，它将在发出的值满足某些要求之后完成观察者，从而使它成为观察者的最后一项。目前，我的代码只是在一个可以观察到的值上使用takeUntil()，当来自源的值满足了标准时就会发出。
const source: Observable<any> = nonCompletingStream();
const conditionSatisfied: Observable<boolean> = source.pipe(
map(val => meetsCriteriaToBeFinalValue(val)),
skip
浏览 0提问于2019-01-18得票数 2
这个问题似乎超出了我的能力范围。我已经正确地构建了存储，现在我无法从存储中检索当前元素。
showDialog(id_clicked) {
this.id = id_clicked
this.store.dispatch(new action.loadExampleAction(this.id));
this.observable$ = this.store.select(selectors.getObject);
}
在这个简单的示例中，每当单击按钮时，我都会更新"observable“，使用当前选定的id分派新的操作。在控制台中显示this.observab
浏览 0提问于2020-07-31得票数 2
我正在学习一些Reactivejs教程，我正在用rx.js实现自动完成。
它看起来很棒，但我想知道如何实现一些功能。看看这段代码：
const $title = $('#title'); // input element
const $results = $('#result');
Rx.Observable.fromEvent($title, 'keyup')
.map(event => event.target.value.trim())
.distinctUntilChanged()
.debounceTim
浏览 1提问于2017-05-21得票数 3
回答已采纳
我有一段代码，它的目的是积累特性，因为它们是在特性$$可观察的基础上添加的。
private features$$: BehaviorSubject<Observable<Feature<FeatureConfig>>> = new BehaviorSubject<
Observable<Feature<FeatureConfig>>
>(of());
private features$: Observable<Feature<FeatureConfig>[]> = thi
浏览 3提问于2022-02-11得票数 1
回答已采纳
为什么下面的代码不起作用？
var observable = Observable.Create<int>(o => new Action(() =>
{
o.OnNext(0);
o.OnCompleted();
}));
foreach (var item in observable.ToEnumerable()) // <-- Here blocks
{
// Never get here
}
以下两种方法都不起作用：
Task.Run(() =>
{
foreach (var item in observable.T
浏览 8提问于2013-08-13得票数 0
回答已采纳
2回答
我阅读了大量关于在角中不订阅Observable的帖子，它们都提到了只在组件/指令中这样做的必要性。
所以我的问题是：
我们是否需要在非组件/指令类中从Observable中取消订阅，例如在服务中，或者基本上包含对Observable的订阅的任何其他类中？
浏览 6提问于2017-09-26得票数 4
回答已采纳
1回答
我有一个函数，它本质上接受一个DOM元素，查找一些标识性的东西，(理论上)返回一个Observable，其中包含前面提到的标识性的东西--withLatestFrom和其他一些东西。
我的问题是，返回的Observable并不会发出任何东西，即使primaries$和highlights$是用do制作的。
如果解释得很糟糕，我非常抱歉，我对ReactiveX/RxJS非常陌生，我正在尽我最大的努力；如果您需要进一步的信息，请询问。
function randomFunction(element) {
// Create an observable sequence of text node
浏览 4提问于2016-08-30得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
