据我所知,.toRx().subscribe(…)函数用于接收消息,而.next()函数用于broADCAST消息

在这个plnkr(http://plnkr.co/edit/MT3xOB?p=info)中,从一个似乎最初从模板定义/派生的数据对象中调用.toRx().subscribe(…)函数：

@Component({
  selector : 'child-cmp',template : '',inputs : ['data']
})
class ChildCmp {
  afterViewInit() {
    this.data.toRx().subscribe((data) => {
      console.log('New data has arrived!',data);
    });
  }
}

在这个plnkr(http://plnkr.co/edit/rNdInA?p=preview)中,从evt对象及其发射器函数(源自注入组件构造函数的Service)调用.toRx().subscribe(…)函数.

@Component({
  selector : 'parent-cmp',template : ''
})
class ParentCmp {
  constructor(evt: EventService) {
    evt.emitter.subscribe((data) => 
      console.log("I'm the parent cmp and I got this data",data));
  }
}

是否可以在服务本身的功能中进行broADCAST,同时,组件可以接收消息而不依赖于返回的服务对象或模板数据对象来链接其.toRX() .subscribe(…)函数调用？

import {Injectable,EventEmitter} from 'angular2/angular2';
@Injectable()
export class DataService {
    items:Array<any>;
    dispatcher: EventEmitter = new EventEmitter();
    constructor() {
        this.items = [
            { name: 'AAAA' },{ name: 'BBBB' },{ name: 'CCCC' }
        ];
    }
    getItems() {
        return this.items;
    }
    sendItems() {
        this.dispatcher.next( this.items );
    } 
}
export var DATA_BINDINGS: Array<any> = [
    DataService
];


@Component({
    selector: 'rabble'
})
@View({
    ...
})
export class Rabble {

    items       : Array<any>;

    constructor( public dataService  : DataService) { 

        console.log('this.routeParam',this.dataService.getItems());
    }

    afterViewInit() {
        this.???.toRx().subscribe((data) => {
            console.log('New item data has arrived!',data);
        });
    }

    handleClick() {
        this.dataService.sendItems();
    }
}

更新至2.0稳定：EventEmitter现在仅用于组件通信.这是Subjects和ReplaySubjects的更好用途.我已将示例更新为2.0代码.

更新到BETA 1：您不再需要在发射器上调用.toRx(),因此我正在更新代码以匹配并向unSubscribe添加示例.

所以现在(Alpha 45)eventEmitter有toRx()方法返回一个RxJS SUBJECT

你可以谷歌搜索一下这是什么以及你可以用它做什么,但这是你实际上在搞乱的东西.当您调用toRx()时,它只返回eventEmitter中的内部主题,因此您可以在服务构造函数中执行此操作.

然后我将您想要进行广播的功能添加到事件服务中

class EventService {
  //Could be regular Subject but I like how ReplaySubject will send the last item when a new subscriber joins
  emitter: ReplaySubject<any> = new ReplaySubject(1);
  constructor() {

  }
  doSomething(data){
    this.emitter.next(data);
  }
}

然后在您的组件中订阅发射器

class ParentCmp {
  myData: any;
  constructor(private evt: EventService) {
    //rx emitter
    this.evt.emitter.subscribe((data) => {
      this.myData = data;
      console.log("I'm the parent cmp and I got this data",data));
    }
  }
}

这是一个扩展的类,内置取消订阅(配置)

export class ParentCmp implements OnDestroy {
  myData: any;
  subscription: any;
  constructor(evt: EventService) {
    //rx emitter
    this.subscription = evt.emitter.subscribe((data) => {
      this.myData = data;
      console.log("I'm the parent cmp and I got this data",data));
    }
  }
  ngOnDestroy() {
    this.subscription.dispose();
  }
}

我对你的上一个问题感到有些困惑,但想到“接收消息”一词.你必须听一些东西,以便订阅方法做什么和需要什么.

很酷的事情现在你可以在整个地方调用那个observable(即使在其他服务中),IMO是组件之间通信的最佳方式.他们不需要知道他们在树中的位置,也不需要关心其他组件是否存在或正在倾听.

我用我的方式工作HERE分叉你的Plunker
(仍在Alpha45上)

RxJs source and info on subject

Angular2 source and info on the subject inside the eventEmitter

当我使用Observable.Interval来执行UI的http刷新时,如果间隔太快,它会锁定UI上的按钮.按钮没有注册点击,似乎是一个时间问题.如果我增加时间并因此错过了获取调用按钮的工作,但数据在更新时会延迟.

间隔

this.dataSub = Observable.interval(1000).subscribe(x => {
  this.getData();
})

的getData

getData(): void {
     this.dataService.getData()
         .subscribe(
              data => this.data = data,error => console.log("Error HTTP Get Service" + this.data),() => {});
 }

有没有最好的做法,或者我缺少的东西,我应该做的是刷新UI而不是锁定按钮

理论

通常,您应该尽可能避免明确订阅observable.相反,使用所有的运算符(是的,找出正确的运算符可能非常棘手)可以将所有源/输入可观察量组合成一个或多个在异步管道视图中使用的可观察对象.

Don’t forget that JS is single threaded (aside from workers). Your UI and most Angular code has to share the one thread,so long running JS locks up your UI.

这有三个主要好处：

>几乎不可能导致内存泄漏.如果您不记得在ngOnDestroy()挂钩中总是取消订阅,或者当您不再关心它时,则无论何时手动订阅都会产生内存泄漏.异步管道在使用它的组件/元素被破坏时将正确取消订阅 – 您无需担心它.
>减少工作量.使用switchMap(),switchLatest()等运算符,您可以取消和清除超级HTTP调用和其他昂贵的操作,甚至可以在它们启动之前停止它们(如果不再需要它们).不要做比你必须做的更多.这通常意味着更改检测不必运行太多,这意味着更好的性能.
>清洁代码.较少的成员变量,更像功能的代码.是的,在学习Rx时可能会有点难以理解,但它会变得更容易.

在实践中

考虑到所有这些,您如何将其应用于您的代码？

您可能没有注意到的一件事(很多人不是),如果你的DataService.getData()方法是这样的：

getData(): Observable<MyData[]> {
    return this.http.get('http://some.url/data').map(res => res.json());
}

然后,每次订阅Http服务创建的observable时,都会发出新请求.这就是你想要的,但是你不想要的是在新的请求发出后立即处理任何先前请求的结果.

因此,您可以使用控制器中的类似内容,使用最新请求中的最新数据组成一个observable：

ngOnInit() {
     // (I follow a convention where observable properties end in $)
     this.data$= Observable.interval(1000).flatMapLatest(() => {
         return this.dataService.getData();
     });
 }

没有订阅,只是一个已创建的可观察对象.然后在您的视图中,只需使用带有数据$属性的异步管道即可.

例如：

<ul *ngFor="let d of (data$| async); trackBy: d?.id">
    <li>{{d.name}}</li>
</ul>

我在组件树的某处创建了一个带有500毫秒的Observable.interval并订阅了它.该组件没有输入或输出属性.该间隔在每次发送滴答时从根组件触发更改检测.这导致我的应用程序中的大量开销,这是不需要的.我没有找到任何关于该行为的文档.

是否可以关闭由此Observable引起的更改检测？

编辑：添加代码

以下代码演示了我想要做的事情.我按照Günter的建议将区间放在Angular区域之外,但现在对阵列的修改不会在模板中发布.有没有办法在不触发更改检测的情况下更新模板？

import {NotificationList} from "./NotificationList";
import {Notification} from "./Notification";
import {Component,OnDestroy,ChangeDetectorRef,ngzone} from "@angular/core";
import { Subscription } from 'rxjs/Subscription';
import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/interval';

class Timednotification {
    notification: Notification;
    remainingTime: number;
}

@Component({
    selector: "notifications",template: `
        <ul>
            <li *ngFor="let notification of notifications">notification.message</li>
        </ul>
    `
})
export class NotificationComponent implements OnDestroy {
    notifications: Array<Timednotification> = [];
    private subscription: Subscription;
    private timer: Subscription = null;
    private delay: number = 2000;
    private tickDelay: number = 500;

    constructor(notificationQueue: NotificationList,private zone: ngzone) {
        this.subscription = notificationQueue.getobservable().subscribe(notification => this.onNotification(notification));
        this.zone.runOutsideAngular(() => {this.timer = Observable.interval(500).subscribe(x => this.onTimer())});
    }

    private onTimer(): void {
        if(this.notifications.length == 0) {
            return;
        }
        let remainingNotifications: Array<Timednotification> = [];
        for(let index in this.notifications) {
            let timednotification = this.notifications[index];
            timednotification.remainingTime -= this.tickDelay;
            if(timednotification.remainingTime <= 0) {
                continue;
            }
            remainingNotifications.push(timednotification);
        }
        this.notifications = remainingNotifications;
    }

    private onNotification(notification: Notification): void {
        let timednotification = new Timednotification();
        timednotification.notification = notification;
        timednotification.remainingTime = this.delay;
        this.notifications.push(timednotification);
    }

    ngOnDestroy(): void {
        this.subscription.unsubscribe();
        if(this.timer !== null) {
            this.timer.unsubscribe();
        }
    }
}

您可以使用ChangeDetectionStrategy.OnPush将组件关闭.

每个事件都会导致更改检测运行(和setTimeout以及ngzone所涵盖的任何其他异步API).

如果您使用OnPush,那么只会更改来自observables订阅的输入和事件|异步导致变化检测.