这是对 Observable 进行排队的合理方式吗？

Question

我对 rxjs 比较陌生，angular。

我有一个需要检索的 projectRev，然后我想检索它的父项。 API 指示两次调用...

我已经阅读了很多关于排队 Observable 的 SO 帖子，但它们都是微不足道的，我迷失了将微不足道的例子转换成适用于我的东西。

这里有一些有用的东西...

第一个 observable 完成后，调用第二个。简单吗？

我的问题：这种方法存在缺陷是否有技术原因。再次......它有效。至少在没有压力的开发机器上....

getItems() {
  console.log('get project rev');
  let prId = this.activatedRoute.snapshot.params.id;
  console.log(prId);

  this.projrevSvc.getItemById(prId).subscribe(
    (data) => {
      console.log('this was fetched :  ' + data);
      this.myItem = data;
      this.getProj();

    });
} //getItems

getProj() {
  console.log('pr.pid ' + this.myItem.ProjectId);
  this.projectSvc.getItemById(this.myItem.ProjectId).subscribe(
    (data) => {
      console.log('this project was fetched :  ' + data);
      this.myProject = data[0];
      console.log('prid ' + this.myItem.id)
      console.log(this.myProject);
    });
}

Answer 1

您需要查看这篇文章以获取更多信息https://blog.angular-university.io/rxjs-higher-order-mapping/

但简而言之，您可以使用类似的东西

getElements() {
  let prId = this.activatedRoute.snapshot.params.id;
  this.projrevSvc.getItemById(prId)
    .pipe(switchMap(response => this.projectSvc.getItemById(response.ProjectId)))
    .subscribe((nestedReponse) => this.myItem = nestedResponse);
}

Answer 2

您的代码有效。没有人可以否认它会奏效。如果路线经常触发，您就会遇到问题，因为如果路线发生变化并且第一次调用 getProj 返回 AFTER 第二次调用 getProj ，那么您将拥有陈旧的数据。所以，虽然它有效，但它不是失败证明。

以下使用 switchMap 以便后续调用被取消，我们绝不会冒数据过时的风险。

this.projrevSvc.getItemById(prId)
  .pipe(
    switchMap(item => {
      return this.projectSvc.getItemById(item.ProjectId)
        .pipe(
          map(project => ([item, project])
        );
    })
  ).subscribe(([item, project]) => console.log(item, project))

然后...您应该通过将 params.id 作为可观察对象来使流完全响应...请参阅此处：

this.activatedRoute.params
  .pipe(
    pluck('id'), 
    switchMap(id => this.projrevSvc.getItemById(id))
    switchMap(item => {
      return this.projectSvc.getItemById(item.ProjectId)
        .pipe(
          map(project => ([item, project])
        );
    })
  ).subscribe(([item, project]) => console.log(item, project))

然后感觉很好，因为您编写了一些完全响应式代码来响应路由器中状态的变化。很漂亮。

Answer 3

is there a TECHNICAL reason this approach is flawed

是的。正如@frosty 指出的那样，当多次执行 getItems() 时，您可能运行进入竞争状态，因为您正在存储数据“ 流之外 ”， this.myItem 的状态取决于您的 http 请求的顺序 return.

虽然这个可能大部分时间都有效，但它并不是完全确定的。

I am getting lost converting the trivial example into something that applies to me

我明白了。 RxJS 很难...起初 :-)

对我变得精通有很大帮助的一件事是意识到：

Observable 本身很无聊
RxJS 使与他们合作变得有价值

这是因为有太多的运算符和静态函数可以让您轻松创建具有明确行为的可观察源。

observable 本质上有两个特征：what 和 when

它发出的数据的形状是什么？
它什么时候会发出这个数据？

您可以将 observables 分解成更小的部分，这使得理解和调试更加容易！

让我们以您的初始示例代码为例（注意：为了让未来的读者清楚起见，我已将 projectRev 重命名为 item）：

export class SomeComponent  {
  
  public myItem    : Item;
  public myProject : Project;

  ngOnInit() {
    this.getItem();
  }

  getItem() {
    let itemId = this.activatedRoute.snapshot.params.id;

    this.itemSvc.getItemById(itemId).subscribe(
      data => {
        this.myItem = data;
        this.getProject();
      }
    );
  }

  getProject() {
    this.projectSvc.getProjectById(this.myItem.ProjectId).subscribe(
      data => this.myProject = data
    );
  }

}

让我们设计一个单独的 Observable，它可以准确地在您需要的时候发出您想要的数据！

提前想到这一点会让生活变得更轻松。

举个例子，假设您要发出一个 Item 及其父级 Project。所以，

what : Item 附加了父 Project 的对象
when ：只要源 item 发生变化（id 不同）

为此，我们可以将所有单独的部分定义为单独的可观察对象。 Angular 提供路由参数作为一个可观察对象，因此与其使用 .snapshot 表示某一时刻的状态，不如定义一个 itemId$ 可观察对象，它将在 时发出 参数更改：

this.itemId$ = this.activatedRoute.params.pipe(pluck('id'));

我们也将 myItem 定义为可观察的。我们希望 myItem$ 发出当前的 Item (what)，每当 id 路由参数改变时 (when):

this.myItem$ = this.itemId$.pipe(
  switchMap(itemId => this.itemSvc.getItemById(itemId))
);

起初，switchMap 可能看起来很混乱（我知道这是给我的）。这是它的作用：

它在内部订阅一个可观察源并发出它的排放物
每次收到新的发射时，它将停止收听以前的来源并订阅新的来源
在您的情况下，我们提供了一个函数，该函数从 itemId$ 和 return 接收到的发射是一个可观察的。这个 observable 是对 this.itemSvc.getItemsById()

因此，希望您能看到每当 itemId$ 发出一个 id 时，myItem$ 将发出 itemSvc.getItemById() 的结果，即 Item 对象。

注意，没有订阅（这由 switchMap 内部处理）。请注意，无需将结果存储在单独的局部变量 this.myItem 中，这是您可能出现竞争条件的原因。

接下来，让我们定义一个 Observable，当一个新的 Item 被发射（当）：

为了冗长：

this.myItemWithProject$ = this.myItem$.pipe(
  switchMap(item => this.projectSvc.getProjectById(item.ProjectId).pipe(
    map(project => ({ ...item, project }))
  ))
);

这里我们将 myItemWithProject$ 定义为在 myItem$ 发出时开始的可观察对象，然后使用我们的新朋友 switchMap 进行调用以获取父对象 Project。然后我们使用 map 来简单地 return Item 对象的一个副本和一个额外的 project 属性.

这里有一个 StackBlitz 完全显示了这一点。

也许你不想要一个单一的组合对象，你显然可以按照你想要的方式塑造数据，也许一个单一的 ViewModel 对象具有 item 和 project 属性.这实际上在 Angular 中很常见：

combineLatest 是一个很好的运算符来为你处理这个问题：

public vm$ : Observable<ViewModel> = combineLatest({ 
 item    : this.myItem$, 
 project : this.myProject$}
);

这种方法允许您在模板中使用单个 observable 和单个 async 管道来解包它：

<div *ngIf ="vm$ | async as vm">

  <h2> My Item </h2>
  <p> {{ vm.item | json }} </p>

  <h2> My Project </h2>
  <p> {{ vm.project | json }} </p>

</div>

随着您的组件变得越来越复杂，您可以简单地向 vm$:

添加更多源

public vm$ : Observable<ViewModel> = combineLatest({ 
 item    : this.myItem$, 
 project : this.myProject$,
 source1 : this.source1$,
 source2 : this.source2$
});

StackBlitz #2

让它“始终可见”可以让事情变得非常简洁和整洁。但是，它要求您了解操作员实际在做什么（什么时候）。

通常没有必要将数据存储在可观察流之外。我发现当我们寻求解决方案时，是因为我们还没有完全理解 rxjs 提供给我们的所有运算符。

Am i right in thinking i should be doing concatMap instead of switchMap

高阶映射运算符之间的区别仅在它们接收到多个发射时才会发挥作用。这是因为它们都订阅了内部资源并排放了它们的排放物。不同之处在于它们在当前源完成之前收到新发射时使用的策略：

switchMap 将源“切换”为仅从最近的源发出。
exhaustMap 将忽略新的发射直到当前源完成，因此它只从第一个源发射。
mergeMap 将从所有新旧来源发出。
concatMap 实际上只是 mergeMap 的一个特例，它一次只允许一个并发源，但最终会从所有源发出

所以，在你的情况下，我认为 switchMap 是合适的，因为如果 id 发生变化，你不再关心使用旧 id 的对象的发射。

Answer 4

好的。非常感谢 bizzyBob 让我走上正轨这花了我 5 天和 3 SO 来弄清楚。我在试图解决这个问题的几个概念上获得了很多清晰度。

我希望这可能对其他人有所帮助... 如果有更好的地方，请随时发表评论...

此示例假设您的服务有效...这里是 projectRev 方法，项目方法几乎相同

   getItemById(id: string): Observable< ProjectRev> {
     console.log('inside proj rev getItemByid..');
       const url = `${this.serviceUrl}/${id}`;
       return this.http.get< ProjectRev>(url, httpOptions)
           .pipe(
             catchError(err =>
               this.handleError(`getItemById=${id}`, err))
       );

   }

这是我的组件的经过编辑的列表

    import { Component, OnInit, AfterViewInit, ElementRef, ViewChild } from '@angular/core';

    import { Location } from '@angular/common';
    import { ChangeDetectorRef } from '@angular/core';
    import { Observable,forkJoin, combineLatest } from 'rxjs';
    import { distinctUntilKeyChanged, pluck
          , switchMap ,tap, map} from 'rxjs/operators';
    import { ActivatedRoute } from '@angular/router';
    
    import { Project } from '../project';
    import { ProjectRev} from '../project-rev';
    import { ProjectService } from '../services/project.service';
    import { ProjectRevService } from '../services/project-rev.service';
    
    // this is the one thing I could not get to work. 
    //i could never get the type of the combineLatest() function to return a VM
    interface  VM{
        projectRev : ProjectRev;
        project : Project;
    }
    
    @Component({
      selector: 'app-project-rev-detail',
      templateUrl: './project-rev-detail.component.html',
      styleUrls: ['./project-rev-detail.component.css']
    })
    export class ProjectRevDetailComponent implements OnInit  {
       
       //THESE ARE ALL **OBSERVABLES**
       //One of the hard lessons i learned doing this is you have to be
       //cognizant of  types. You have class types you designed, then 
       // OBSERVABLES, and 
       //  SUBSCRIPTIONS. obviously a<>b<>c , but its easy to overlook
       // I was using ATOM at the start of this exercise.  VSCODE does a 
       // much better job pointing out TYPE mismatches. 
       private projectRevId$ : Observable<string>;
       private myItem$ : Observable<ProjectRev>;
       private myProject$ : Observable<Project>;
    
        //note that this is of type any. 
       public vm$ : Observable<any>;
     
      constructor( private projrevSvc : ProjectRevService
             , private projectSvc : ProjectService
              , private activatedRoute: ActivatedRoute
           ) {
              
            //you are not really supposed to put anything here, if you believe the internet... 
            
           }
    


      //be sure to indicate that your class implements onInit.
      // you also have to import it from '@angular/core'
      ngOnInit(): void {
    

          /*
          This is where the magic happens. 
          We are going to assign the 4 observables
          Except for the first one, each one does a pipe() off the previous

          Couple things i didnt realize about pipe

> A Pipeable Operator is essentially a pure function which takes one
> Observable as input and generates another Observable as output.
> Subscribing to the output Observable will also subscribe to the input
> Observable.
https://rxjs.dev/guide/operators

          pipe() is called on an observable, its a method on an observable
          pipe() returns a NEW observable, it does not change the input observable
          pipe() has a parm list for its signature. Thats why the syntax looks odd. 
       in parm, you can call any number of map operators

          Each operator looks something like this: 
                     map( (proj) =>  { console.log( proj) } )

          the parameter proj can be named whatever. 
          I like to read it this way:  I got back 'proj' , so i FatArrow  <<my stuff>>
          Fat Arrow can be followed by a single statement, no semicolon necessary
          Fat Arrow can also be followed by curlies containing multiple statements. 
              Semicolons are required here.
         
           ** Make sure you put the returned value(s) in parens before the FatArrow
           ** Make sure you get the closing paren on the operator. Its best to always count parens on each line. They all have to be enclosed inside the pipe() parens.  Its very , very easy to mess it up. good luck!


          If you want to test your chain 1 at a time, be sure and subscribe to the last one you are looking at. 
          I have left my subscribes commented, but in place as an example 
 

*/




        //**
        //** Get the id from the active route
        //**
        this.projectRevId$ = this.activatedRoute.params.pipe(
            pluck('id')
            ,tap( id => {  console.log('pRevId',id);})
            );
            //).subscribe( (id)=> console.log('the url id is :',id));
       



        //**
        //** Get the item (projectRev) using the ID from the URL
        //**
        //**  switchMap
        //**   https://rxjs.dev/api/operators/switchMap
        //**   still not sure switchMap is the best operator, but its working
        //**
            this.myItem$ = this.projectRevId$.pipe(
                tap( item => {  console.log('inside projrev pipe',item);})
                ,   switchMap( (itemId : string) => this.projrevSvc.getItemById(  itemId   ))
                , tap( item => console.log('rev retrieved',item))
            );
            // ).subscribe( (pr) => console.log('projecRev=',pr));
    



        //**
        //** Get the project using the just retrieved projectRev.projectid value
        //** https://rxjs.dev/api/operators/distinctUntilKeyChanged - nice....
        //** notice i use 'pr' in switchMap and 'item' in tap
        //**  i was trying to test what matters syntax wise...
        //**  readability wise, consistency and 'obvious' is better
            this.myProject$ = this.myItem$.pipe(
                tap( (prev : ProjectRev) =>{  console.log('inside getProj pipe');})
                ,distinctUntilKeyChanged('ProjectId')
                ,switchMap((pr) => 
                        this.projectSvc.getItemById(pr.ProjectId))
                , tap( item => console.log('proj retrieved',item))
             );  
            //  ).subscribe( (p ) => console.log('projec =',p ));
    


        //**
        //** And finally, combine them retrieved 
        //** https://rxjs.dev/api/index/function/combineLatest
        //**
        //** NOTE: the 2 OBSERVABLES are passed in as a single array 
        //**
        //** in the map(), i used better names, but note they dont match 
        //** what I used before...
        //** the map is from this SO
        //**
        //**  this map is important when we move on to the HTML


            this.vm$ = combineLatest(
                  [ 
                  this.myItem$,
                  this.myProject$
                  ]
              ).pipe(
                 map( ([projectRev,project ]) => ({projectRev, project}))  
              );
              
              //we DO NOT need a SUBSCRIBE because we 
              //  are using the async pipe on the outer 
              //  div in the HTML and that does a subscribe
              // IF YOU SUBSCRIBE HERE, YOU WILL DO EVERYTHING TWICE

      }
    }

最后，我们可以在 HTML 模板中使用它这是一篇关于正确使用 ngIf 的好文章... https://blog.angular-university.io/angular-reactive-templates/

嗯，async 的末尾没有 H。
ngIf 有大写字母 I
vm$ 需要是可观察的
vm1 不需要在你的 class
中定义
async 管道对命名的 Observable 进行订阅。容易忘记...

带 {{vm1}} 的 div 将向您展示 vm1 实际上是什么。它可能不是你想的那样。很好的调试，可以弄清楚为什么什么都没有出现。

   <ng-template #loading>
     <div class="col-xs-8" >loading</div>
 </ng-template>  
 <div class="col-xs-8"  *ngIf="vm$ | async as vm1; else loading">
   <div> {{vm1 }} </div>
   <div ><a href="#" >{{  vm1.project.Name     }} </a> >
           <a href="#"> Rev. {{ vm1.projectRev.RevisionNumber }} - {{vm1.projectRev.Name}} </a>
     </div>
 </div>

瞧 - 我的面包屑以完全异步的方式显示！！！！我知道这看起来很疯狂，但现在我拥有了完成页面其余部分所需的所有数据。即使我确实需要在第 3 个调用中进行混合，我觉得我已经很好地掌握了这项技术，只需几分钟即可混合另一个数据源。

正在重读其他 2 个答案... Frosty 将它们链接在一起，提高了简洁性

BizzyBob 实际上在声明部分定义并初始化了所有可观察对象。

完成同一件事的方法有很多种。我认为这就是通过对不同的 SO 答案进行采样来合成有效解决方案变得困难的部分原因。

这是对 Observable 进行排队的合理方式吗？

Is this a reasonable way to queue Observables?

rxjs

angular