使用RxJava / Jersey2的异步RestAPI。 线程问题？
我们正在使用反应式编程对REST API进行原型设计。 如图所示，我们保留3层与我们在上一代同步API设计中使用的相同;
http://oi59.tinypic.com/339hhki.jpg
- 使用Jersey2实现的API层将处理请求/反序列化JSON并切换到服务层。
- 服务层实现业务逻辑。使用反应式编程实现（RxJava）
- Dao Layer用于Service Layer的持久化操作。因为我们使用CouchBase，所以这将使用CouchBase RxClient。
根据我的理解，流程如下：
a） HTTP请求到来，Jersery将从“容器线程池”处理RequestThread内的请求/解析JSON /反序列化请求模型。
b）使用Jersey2异步支持，RequestThread将返回到Container Thread Pool，服务层将在Schedulers.computation（）调度程序中执行。
@Path("/resource") public class AsyncUserResource { @GET public void asyncGet(@Suspended final AsyncResponse asyncResponse) { Observable user = userService.getUser(...); //this is executed using Schedulers.computation() inside Service implementation user.subscribe(new Observer() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { //handle error using ExceptionMappers } @Override public void onNext(User user) { asyncResponse.resume(user); }}); } } c） DAO中的任何IO操作都将使用Schedulers.io（）在单独的线程中运行这些长处理操作。
我的问题是：
- 在实现DAO / Services时，我是否应该在实现中隐藏正在使用的调度（Threading）。
例如Dao：
public interface UserDao { public Observable getUser(); }
在实施中，如下指定Schedular是一种好的做法;
public Observable getUser() { Observable ret = Observable.create((subscriber)->{ try { //Do DB call User u = null; subscriber.onNext(u); subscriber.onCompleted(); }catch (Exception e) { subscriber.onError(e); } }); return ret.subscribeOn(Schedulers.io()); }
或者简单地返回Observable是否更好，而上层将相应地使用特定的Schedular？
-
由于DAO主要是入侵io /网络呼叫，我假设应该使用Schedulars.io（）。 服务层一侧的业务逻辑怎么样？ 它们应该在Schedulers.computation（）（事件循环）中执行吗？
-
JVM中有两个线程池。一个是“容器线程池”，另一个是Schedulers.io（）使用的“RxThread池”。 如何配置RxJava的池设置/大小？
1）在RxJava本身中，如果方法需要调度程序，我们创建两个重载：一个没有Scheduler参数，另一个带有它。 然后前者使用合理的默认调度程序委托后者。 这样，API使用者可以选择接受默认值或使用自己的默认值。
2）这取决于你的计算。 如果计算花费相同的时间等待IO，则可以将计算移动到计算调度程序中，从而释放IO中的缓存工作线程以执行更多阻塞。 否则，您可以在同一个调度程序上执行业务逻辑。
3）您目前无法在RxJava中配置池大小。 计算将始终使用Runtime.availableProcessors()和IO将始终充当无限制的缓存线程池。 如果您可以使用事件跳线（意思是：它们保证是串行的，但一个事件可以在线程1上执行，后续在线程2上执行），您可以通过Schedulers.from()使用您自己的ExecutorServices。