class ResultSets {

    public static Observable<ResultSet> queryAllAsObservable(Session session, String query, Object... partitionKeys) {
        List<ResultSetFuture> futures = sendQueries(session, query, partitionKeys);
        //转换为Rx的模式, 将ListenableFuture转换为观察者类型Observable, 最后合并为一个Observable<ResultSet>
        Scheduler scheduler = Schedulers.io();
        List<Observable<ResultSet>> observables = Lists.transform(futures, (ResultSetFuture future) -> Observable.from(future, scheduler));
        return Observable.merge(observables);
    }

}