鸟瞰 Java 并发框架

8. RxJava

• 这与上面的情况类似，唯一的区别是 RxJava 提供了更好的 DSL 可以进行流式编程，下面的例子中没有体现这一点。
• 性能优于 CompletableFuture 处理并行任务。

8.1 何时使用?

如果编码的场景适合异步非阻塞方式，那么可以首选 RxJava 或任何响应式开发库。还具有诸如 back-pressure 之类的附加功能，可以在生产者和消费者之间平衡负载。

int userId = new Random().nextInt(10) + 1; 
ExecutorService executor = CustomThreads.getExecutorService(8); 
 
// I/O 任务 
Observable<String> postsObservable = Observable.just(userId).map(o -> JsonService.getPosts()) 
.subscribeOn(Schedulers.from(executor)); 
Observable<String> commentsObservable = Observable.just(userId).map(o -> JsonService.getComments()) 
.subscribeOn(Schedulers.from(executor)); 
Observable<String> albumsObservable = Observable.just(userId).map(o -> JsonService.getAlbums()) 
.subscribeOn(Schedulers.from(executor)); 
Observable<String> photosObservable = Observable.just(userId).map(o -> JsonService.getPhotos()) 
.subscribeOn(Schedulers.from(executor)); 
 
// 合并来自 /posts 和 /comments API 的响应 
// 作为这个操作的一部分，将执行内存中的一些任务 
Observable<String> postsAndCommentsObservable = Observable 
.zip(postsObservable, commentsObservable, 
(posts, comments) -> ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments)) 
.subscribeOn(Schedulers.from(executor)); 
 
// 合并来自 /albums 和 /photos API 的响应 
// 作为这个操作的一部分，将执行内存中的一些任务 
Observable<String> albumsAndPhotosObservable = Observable 
.zip(albumsObservable, photosObservable, 
(albums, photos) -> ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos)) 
.subscribeOn(Schedulers.from(executor)); 
 
// 构建最终响应 
Observable.zip(postsAndCommentsObservable, albumsAndPhotosObservable, (r1, r2) -> r1 + r2) 
.subscribeOn(Schedulers.from(executor)) 
.subscribe((response) -> asyncResponse.resume(response), e -> asyncResponse.resume("error")); 

（编辑：91站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!