小编eln*_*ino的帖子

RxJava - synchronized块 - 内部源

我想使用synchronized块作为flatMap的源码.但我需要使用此构造进行处理(方法processItem),而不仅仅是在创建内部源时.

每隔5分钟调用一次Observable(例如Observable.interval)

Observable.fromIterable(getSourceData())
.flatMapCompletable(item -> {
  synchronized(LockManager.getInstance().getLockObject(item.id)){
    return processItem(item)
          .subscribeOn(Schedulers.io());
  }
})
Run Code Online (Sandbox Code Playgroud)

我的processsItem方法如下所示:

public Completable processItem(Item item){
  return mApiSource.getItemById(item.id)
    .flatMap(item -> 
      mItemRepository.replace(item)
    )
    .toCompletable();
}
Run Code Online (Sandbox Code Playgroud)

两个内部方法都返回Single.

它是从服务器定期更新的方法的一部分.我需要序列化processItem方法调用(从服务器定期同步项目)和修改项目(更新,删除)的方法,这些方法是从其他项目类调用的(项目同步).

主要问题是定期更新可以重写新更新的项目.

其实我用这个解决方案:

用于更新新项目的链:

public Completable updateItem(Item item){
  return Completable.fromAction(() -> {
        synchronized(LockManager.getInstance().getLockObject(item.id)){
          mApiSource.update(item)
          .flatMap(item -> 
            mItemRepository.replace(item)
          )
          .toCompletable()
          .blockingAwait();
        }
      })
      .subscribeOn(Schedulers.io())
}
Run Code Online (Sandbox Code Playgroud)

定期更新链:

Observable.fromIterable(getSourceData())
    .flatMapCompletable(item -> {
      Completable.fromAction(() ->{
        synchronized(LockManager.getInstance().getLockObject(item.id)){
          processItem(item).blockingAwait();
        }
      })
      .subscribeOn(Schedulers.io())
    });
Run Code Online (Sandbox Code Playgroud)

我知道RxJava解决方案并不清楚.

你知道更好的解决方案吗?

synchronized synchronized-block thread-synchronization rx-java2

5
推荐指数
0
解决办法
248
查看次数

Java中的单线程异步处理

我对Java线程有疑问。当我仅使用主线程时,可以异步运行两个任务吗?我读了这篇文章:异步与同步执行,这实际上意味着什么? 第二个答案是可能的情况,其中只有一个线程(而不是主线程和另一个线程)并且任务被异步执行。我认为这在Java中是不可能的,因为仅存在一个控制流。我认为在Java中,每个任务都需要自己的线程来异步执行任务。

java asynchronous

4
推荐指数
1
解决办法
1948
查看次数