标签: reactive-programming

我正在使用RxJava链接异步操作,我想将一些变量传递到下游:

Observable
   .from(modifications)
   .flatmap( (data1) -> { return op1(data1); })
   ...
   .flatmap( (data2) -> { 
       // How to access data1 here ?
       return op2(data2);
   })

这似乎是一种常见的模式,但我无法找到有关它的信息.

我最近一直在对反应式编程进行一些研究,我发现很难找到Reactive和Functional -Reactive 之间差异的良好描述.

是否只是使用功能方法/范例实现反应式编程而不是使用声明式或OO范式？

我回顾了以下SO问题: 什么是热和冷可观测量？

总结一下:

• 当一个cold observable有一个观察者使用它时,它会发出它的值,即观察者接收的值序列与订阅时间无关.所有观察者都将使用相同的值序列.
• 热的observable独立于其订阅发出值,即观察者接收的值是订阅时间的函数.

然而,我觉得热和冷仍然是混乱的根源.所以这是我的问题:

• 默认情况下所有rx可观察量是否都是冷的(主题除外)？

  我经常读到事件是热观察的典型隐喻,但我也读到这Rx.fromEvent(input, 'click')是一个冷可观察的(？).

• 是否有什么/哪些Rx运算符将冷观测值转换为热观测值(除了publish和之外share)？

  例如,它如何与Rx运算符一起使用withLatestFrom？让我们cold$成为一个冷酷的观察者.会sth$.withLatestFrom(cold$,...)是一个热门观察？

  或者,如果我不sth1$.withLatestFrom(cold$,...), sth2$.withLatestFrom(cold$,...)和订阅sth1和sth2,将我总是看到两个相同的值sth？

• 我认为Rx.fromEvent会产生冷的可观测量,但事实并非如此,正如其中一个答案所述.但是,我仍然对此行为感到困惑:codepen.io/anon/pen/NqQMJR?editors=101.不同的订阅从同一个observable获得不同的值.click事件不是共享的吗？

在将Clojure的core.async与所谓的Reactive Extensions(Rx)和FRP进行比较时,我似乎有点困惑.他们似乎解决了类似的异步 - 长期问题,所以我想知道什么是主要差异,在什么情况下优先于另一个.有人可以解释一下吗？

编辑:为了鼓励更深入的答案,我想让问题更具体:

1. Core.async允许我编写同步代码.但是据我所知,FRP只需要一级嵌套回调(处理逻辑的所有函数都作为参数传递给FRP API).这似乎两种方法都不需要回调金字塔.确实,在JS中我必须写function() {...}多次,但主要问题,嵌套回调,也在FRP中消失了.我做对了吗？

2. " FRP通过控制流完善了信息的沟通"你能(某人)请给出更具体的解释吗？

3. 我不能像通过频道那样绕过FRP的可观察端点吗？

总的来说,我理解历史上两种方法的来源,我在这两种方法中都尝试了很少的教程.然而,我似乎被差异的非显而易见性"瘫痪"了.是否有一些代码难以在其中一种中编写而使用另一种代码很容易？那架构原因是什么？

我正在学习RxJS和Angular 2.假设我有一个带有多个异步函数调用的promise链,它取决于前一个的结果,如下所示:

var promiseChain = new Promise((resolve, reject) => {
  setTimeout(() => {
    resolve(1);
  }, 1000);
}).then((result) => {
  console.log(result);

  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve(result + 2);
    }, 1000);
  });
}).then((result) => {
  console.log(result);

  return new Promise((resolve, reject) => {
      setTimeout(() => {
      resolve(result + 3);
        }, 1000);
  });
});

promiseChain.then((finalResult) => {
  console.log(finalResult);
});

我在不使用promise的情况下单独使用RxJS的尝试产生了以下结果:

var observableChain = Observable.create((observer) => {
  setTimeout(() => {
    observer.next(1);
    observer.complete();
  }, 1000);
}).flatMap((result) => {
  console.log(result);

  return Observable.create((observer) …

我需要一些帮助来实现RxJava中的并行异步调用.我已经选择了一个简单的用例,其中FIRST调用获取(而不是搜索)要显示的产品列表(Tile).随后的电话会出来并获取(A)评论和(B)产品图片

经过几次尝试,我到了这个地方.

 1    Observable<Tile> searchTile = searchServiceClient.getSearchResults(searchTerm);
 2    List<Tile> allTiles = new ArrayList<Tile>();
 3    ClientResponse response = new ClientResponse();

 4    searchTile.parallel(oTile -> {
 5      return oTile.flatMap(t -> {
 6        Observable<Reviews> reviews = reviewsServiceClient.getSellerReviews(t.getSellerId());
 7        Observable<String> imageUrl = reviewsServiceClient.getProductImage(t.getProductId());

 8        return Observable.zip(reviews, imageUrl, (r, u) -> {
 9          t.setReviews(r);
10          t.setImageUrl(u);

11          return t;
12        });

13      });
14    }).subscribe(e -> {
15      allTiles.add((Tile) e);
16    });

第1行:熄灭并取出要显示的产品(平铺)

第4行:我们获取Observable和SHARD的列表以获取评论和imageUrls

谎言6,7:获取Observable评论和Observable url

第8行:最后将2个可观察对象压缩以返回更新的Observable

第15行:最后第15行整理所有要在集合中显示的单个产品,这些产品可以返回到调用层

虽然Observable已被分片,但在我们的测试中运行了4个不同的线程; 获取评论和图像似乎是一个接一个.我怀疑第8行的zip步骤基本上导致了2个observables(review和url)的顺序调用.

这个小组是否有任何关于平行获取reiews和图片网址的建议.实质上,上面附带的瀑布图应该看起来更垂直堆叠.对评论和图像的调用应该是并行的

谢谢anand raman

给出一个汽车列表(List<Car> cars),我可以这样做:

Observable.just(cars); //returns an Observable that emits one List<Car>
Observable.from(cars); //returns an Observable that emits a squence of Car

有没有办法可以从Observable一个List<Car>到一个序列Observable<Car>？

像from没有参数的东西

Obserable.just(cars).from()

现在我几次读到反应式编程的闪亮的文档,但我不能正确理解之间的差异observeEvent和eventReactive.

文件说:

只要您想执行响应事件的操作,请使用observeEvent.(请注意,"重新计算值"通常不会被视为执行操作 - 请参阅eventReactive.)

....

使用eventReactive创建仅在响应事件时更新的计算值.这就像一个普通的反应式表达式,除了它忽略了来自其反应依赖的所有常见的失效;

在我试过的所有情况下,我发现使用observeEvent和之间没有区别eventReactive(无论我使用哪种功能,代码都能正常工作,对性能没有明显的影响).

你能帮我弄清楚两者之间的真正区别吗？理想情况下,我想举几个例子说明它们何时可以互换,哪一个observeEvent可以工作但不能eventReactive,反之亦然.

我正在尝试打印使用中的http调用结果Angularrxjs

请考虑以下代码

import { Component, Injectable, OnInit } from '@angular/core';
import { Http, HTTP_PROVIDERS } from '@angular/http';
import 'rxjs/Rx';

@Injectable()
class myHTTPService {
  constructor(private http: Http) {}

  configEndPoint: string = '/my_url/get_config';

  getConfig() {

    return this.http
      .get(this.configEndPoint)
      .map(res => res.json());
  }
}

@Component({
    selector: 'my-app',
    templateUrl: './myTemplate',
    providers: [HTTP_PROVIDERS, myHTTPService],


})
export class AppComponent implements OnInit {

    constructor(private myService: myHTTPService) { }

    ngOnInit() {
      console.log(this.myService.getConfig());
    }
}

每当我试图打印出getconfig它的结果总是返回

Observable {_isScalar: false, source: Observable, operator: MapOperator} …

在我的应用程序中我有类似的东西:

this._personService.getName(id)
      .concat(this._documentService.getDocument())
      .subscribe((response) => {
                  console.log(response)
                  this.showForm()
       });

 //Output: 
 // [getnameResult]
 // [getDocumentResult]

 // I want:
 // [getnameResult][getDocumentResult]

然后我得到两个分开的结果,第一个_personService然后是_documentService.如何在调用之前等待两个结果this.showForm()然后操纵每个结果.