标签: subject-observer

我使用共享服务在两个独立组件之间传递数据。当我在共享服务中使用主题时，我无法在订阅组件的html中看到订阅的数据，但订阅后可以在控制台中看到它。而如果我使用Behaviorsubject，效果很好。谁能给我解释一下原因。

共享服务.ts：

  //code with subject (not working)
  private msgStatistics = new Subject<any>();
      msgStatistics$ = this.msgStatistics.asObservable();
      msgStats(message) {
        this.msgStatistics.next(message)
      } 


    //code with BehaviorSubject (working)
    private msgStatistics = new BehaviorSubject<Object>(null);
      msgStatistics$ = this.msgStatistics.asObservable();
      msgStats(message) {
        this.msgStatistics.next(message)
      }

组件2.ts：

this.shared_service.msgStatistics$.subscribe(res => {
        this.message = res
        console.log(this.message) 
})

上面的控制台在这两种情况下都会打印消息的值，但它不会以 html 形式呈现主题。

我正在运行RxJava并创建一个使用onNext()方法来生成数据的主题.我正在使用Spring.

这是我的设置:

@Component
public class SubjectObserver {
    private SerializedSubject<SomeObj, SomeObj> safeSource;
    public SubjectObserver() {
       safeSource = PublishSubject.<SomeObj>create().toSerialized();
       **safeSource.subscribeOn(<my taskthreadExecutor>);**
       **safeSource.observeOn(<my taskthreadExecutor>);** 
       safeSource.subscribe(new Subscriber<AsyncRemoteRequest>() {
          @Override
          public void onNext(AsyncRemoteRequest asyncRemoteRequest) {
            LOGGER.debug("{} invoked.", Thread.currentThread().getName());
            doSomething();
          }
      }
    }
    public void publish(SomeObj myObj) {
        safeSource.onNext(myObj);
    }
}

在RxJava流上生成新数据的方式是通过@Autowire private SubjectObserver subjectObserver 然后调用subjectObserver.publish(newDataObjGenerated)

无论我为subscribeOn()&指定什么observeOn():

• Schedulers.io()
• Schedulers.computation()
• 我的主题
• Schedulers.newThread

其中onNext()的实际工作是在实际调用onNext()主题以生成/生成数据的同一线程上完成的.

它是否正确？如果是这样,我错过了什么？我期待在doSomething()不同的线程上完成.

更新

在我的调用类中,如果我改变了调用publish方法的方式,那么当然会为订阅者分配一个新线程来运行.

taskExecutor.execute(() -> subjectObserver.publish(newlyGeneratedObj)); …

我有一个名为BookService的服务项目。

private books: Subject<Book[]>;

getBookList(skip:number = 0,limit:number = 0): Subject<Book[]> {
  return this.books;
}

addBookToList(book:Book) {
}

此服务可帮助我监视书籍列表中的更改，我还希望能够使用addBookToList函数将单个书籍添加到列表中，但是我不知道如何添加可观察到的单个书籍（主题观察者）是否可以添加一本书

我正在构建一个过滤器，您可以在其中按类别过滤，您可以通过单击类别名称旁边的复选框来选择一个类别。

所以我有一个filterComponent，它包含它自己的过滤器，然后是一个filterService，它有一个类别属性Subject<Array<ICategory>>，这个属性用于将数据传递到productsComponent我订阅类别属性的位置。

当我想使用此模式传递一个简单的字符串时，此逻辑有效，但当我想传递对象数组时，它似乎不起作用。

在我的 filters.component.html 文件中，当复选框值发生变化时，我正在调用一个方法：

<li *ngFor="let category of categories">
        <mat-checkbox (change)="addOrRemoveCategory(category)" [(ngModel)]="category.isChecked">
                {{'category.' + category.Name | translate}}
        </mat-checkbox>
</li>

addOrRemoveCategory 方法实现如下所示：

private addOrRemoveCategory(category: ICategory): void {
    if (!this.chosenCategories.includes(category)) {
        this.add(category);
    } else {
        this.remove(category);
    }
    this.filterService.categories.next(this.chosenCategories);
}

所以无论一个类别发生了什么，被添加或删除，（我在内部修改了selectedCategories数组，我用它的值调用 .next() ），我用更新的数组调用 .next() 。

问题是，当selectedCategories数组为空时，我推送到它，并用它调用 .next() 时，我正确地获取了订阅者函数中的值，但是如果再次执行此操作，并且我有一个2 个元素数组，我调用 .next(this.chosenCategories)，我的订阅者方法没有收到通知。

但是，一旦我使用空数组调用 .next() ，我的订阅者方法就会再次收到通知（因为我已经删除了之前选择的所有类别）。

订阅者方法：

this.categoriesChangeSubscription = this.filterService.categories
            .pipe(
                debounceTime(500),
                distinctUntilChanged()
            )
            .subscribe((categories: Array<ICategory>) => { …