可以在rxjava中转换以下代码吗

Question

例如，我有以下可运行的java代码。

它是关于一个生产者和几个并行的消费者。这些使用者正在运行耗时的作业，并且它们是并行运行的。

我想知道这个用例是否与rx-java匹配，以及如何用rx-java重写它。

public class DemoInJava {
    public static void main(String[] args) {

        final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

        AtomicBoolean done = new AtomicBoolean(false);
        Thread producer = new Thread(() -> {
            int offset = 0;
            int limit = 10;
            while (true) {
                if (queue.isEmpty()) {
                    if (offset < 100) {// there is 100 records in db
                        fetchDataFromDb(offset, limit).forEach(e -> queue.add(e));
                        offset = offset + limit;
                    } else {
                        done.set(true);
                        break; // no more data
                    }
                } else {
                    try {
                        Thread.sleep(100l);// I don't like the idea of sleep, but it seems to be the simplest way.
                    } catch (InterruptedException e) {
                    }
                }
            }
        });

        List<Thread> consumers = IntStream.range(0, 5).boxed().map(c -> new Thread(() ->
        {
            while (true) {
                Integer i = queue.poll();
                if (i != null) {
                    longRunJob(i);
                } else {
                    if (done.get()) {
                        break;
                    } else {
                        try {
                            Thread.sleep(100l);// I don't like the idea of sleep, but it seems to be the simplest way.
                        } catch (InterruptedException e) {
                        }
                    }
                }
            }
        })).collect(Collectors.toList());

        producer.start();
        consumers.forEach(c -> c.start());
    }

    private static List<Integer> fetchDataFromDb(int offset, int limit) {
        return IntStream.range(offset, offset + limit).boxed().collect(Collectors.toList());
    }

    private static void longRunJob(Integer i) {
        System.out.println(Thread.currentThread().getName() + " long run job of " + i);
    }
}

输出为：

....
Thread-1 long run job of 7
Thread-1 long run job of 8
Thread-1 long run job of 9
Thread-4 long run job of 10
Thread-4 long run job of 16
Thread-10 long run job of 14
Thread-5 long run job of 15
Thread-8 long run job of 13
Thread-7 long run job of 12
Thread-9 long run job of 11
Thread-10 long run job of 19
Thread-4 long run job of 18
Thread-3 long run job of 17
....

Answer 1

让我看看..。首先，代码：

package rxtest;

import static io.reactivex.Flowable.generate;
import static io.reactivex.Flowable.just;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import io.reactivex.Emitter;
import io.reactivex.Scheduler;
import io.reactivex.schedulers.Schedulers;

public class Main {

    private static final Scheduler SCHEDULER = Schedulers.from(Executors.newFixedThreadPool(10));

    private static class DatabaseProducer {
        private int offset = 0;
        private int limit = 100;

        void fetchDataFromDb(Emitter<List<Integer>> queue) {
            System.out.println(Thread.currentThread().getName() + " fetching "+offset);
            queue.onNext(IntStream.range(offset, offset + limit).boxed().collect(Collectors.toList()));
            offset += limit;
        }
    }

    public static void main(String[] args) {
        generate(new DatabaseProducer()::fetchDataFromDb)
        .subscribeOn(Schedulers.io())
        .concatMapIterable(list -> list, 1) // 1 call, no prefetch
        .flatMap(item -> 
                just(item)
                .doOnNext(i -> longRunJob(i))
                .subscribeOn(SCHEDULER)
                , 10) // don't subscribe to more than 10 at a time
        .take(1000)
        .blockingSubscribe();
    }

    private static void longRunJob(Integer i) {
        System.out.println(Thread.currentThread().getName() + " long run job of " + i);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

类DatabaseProducer只是值的有状态生产者，因为它需要当前的偏移量。这并不是绝对必要的，因为generate调用可以替换为

        generate(() -> 0, (offset,e) -> {
            e.onNext(IntStream.range(offset, offset + 100).boxed()
                       .collect(Collectors.toList()));
            return offset + 100;
        }, e -> {});

但它的可读性不是很好。

请记住，cocatMap和flatMap可以并将预取和预订阅可观察对象/可流动对象，直到达到实现相关的限制，即使没有空闲线程处理它们-它们只会在调度器中排队。每次调用的数字代表我们希望在concatMap上有-1的限制，因为我们只想在必要时从数据库获取(如果您在这里放2，您可能会过度读取，但管道中的延迟会更少)。

如果您想要进行Cpu限制的计算，那么最好使用Schedulers.computation()，因为它会自动配置为运行JVM的系统的Cpu数量，并且您可以从代码库的其他部分使用它，这样就不会使处理器过载。