关于 rxjs 可观察对象的基本问题——如何“手动”将数据提供给可观察对象？答案

【问题标题】：Basic question on rxjs observables --how to feed data to an observable 'by hand'?关于 rxjs 可观察对象的基本问题——如何“手动”将数据提供给可观察对象？
【发布时间】：2020-04-06 19:09:23
【问题描述】：

我想使用 observable 来向我的 Angular 应用程序的各个部分发出关于“异常状态”的信号，但我意识到我并不真正了解它们是如何工作的。

在下面的代码中，我构建了一个观察者对象，并从中创建了一个可观察对象。我想做的是弄清楚如何在 Observable.create 方法范围之外调用“next”方法，以便我可以将任意事件插入流中。直接在观察者身上调用 next 似乎不是答案。

 var observer = {
      next: function(value) {
        this.myvalue="last value: " + value;
      },
      error: function(error) {
        console.log(error);
      },
      complete: function() {
        console.log('completed');
      },
      myfunction: function() {
        this.myvalue = "Penguins"
      },
      myvalue: ""
    }

   let myobs$ : Observable<any> = Observable.create(function(obs) {
      obs.next("Some foo");

      obs.next("Other foo");
    })

    let foo=myobs$.subscribe((observer)=> {
      console.log("This is the subscription: ", observer)
    })

    setTimeout(function() {
      observer.next("This is a late breaking value");
      console.log(observer.myvalue);
    }, 2000);

  }

此代码产生以下控制台输出：

This is the subscription:  Some foo
This is the subscription:  Other foo
last value: This is a late breaking value

所以看起来直接在观察者对象上调用next（我在底部的超时函数中尝试过）不会在订阅中产生值。

也很清楚，我想我不明白这些东西应该如何工作。了解如果我设置了一个可观察对象并且我想“手动”将数据插入到订阅者接收的流中，我该怎么做？我可以看到你是如何使用鼠标点击或 ajax 请求等事件传播的事情来做到这一点的，但我想做的是创建一个流，当某些有趣的事情发生在不同的地方时，我可以临时提供它我的代码。

【问题讨论】：

标签： angular typescript rxjs observable

【解决方案1】：

您应该阅读on RxJs Subject。这将是你所需要的。来自文档：

Subject 是一种特殊类型的 Observable，它允许将值多播到多个 Observer。主题就像 EventEmitters。

有不同类型的主题。 ReplaySubject、BehaviorSubject、AsyncSubject。我建议您深入研究文档或一些关于如何以及何时使用它们的教程。从您的 observable 中，您还希望发出以前的值和新值。您可以使用the pipable operators that RxJs ships with 进行此类自定义行为。我建议你也阅读这些。 Here is another post on stackoverflow 正是你想要的：

基于该示例：

const subject = new Subject().pipe(
  startWith('Penguins'), // emitting first value to fill-in the buffer
  pairwise(),
  map([previous, current] => {
    return previous + current;
  }),
);

const observer = {
  next: (value) => {
    console.log(value);
  },
  error: (error) => {
    console.log(error);
  },
  complete: () => {
    console.log('completed');
  },
};

// Subjects
const subscriber = subject.subscribe(observer);

subject.next('Some foo');

【讨论】：

【解决方案2】：

简而言之：您不能以这种方式创建可观察对象。您实际上作为subscribe() 方法的参数传递的只是一个在对象状态更改时将调用的函数。此功能（以及其他功能，因为您可以多次订阅）将被存储，因此您必须记住取消订阅。 Observables 只不过是观察者模式的 rxjs 实现。但是，如果有变化，还有另一种方式来发出信号。代替number、string 等变量，您可以使用Subject 或BehaviourSubject 甚至ReplaySubject 来轻松通知状态是否发生变化。 Subject不存储您使用next() 方法传递的值，因此只有在使用next() 之前订阅时才会收到。 BehaviourSubject 存储价值，当您订阅时，您将获得最新价值。 ReplaySubject 与 BehaviourSubject 类似，但它会发出 n 个最新值。

编辑：实际上你可以，但是你必须存储你在回调中传递给Observable.create的函数，你可以在Observable之外使用next发送一些数据。但即便如此，我还是强烈建议您使用Subjects。

【讨论】：