0

例如，我有以下可运行的 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
....
4

1 回答 1

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每个调用上的数字代表我们想要的限制 - 1concatMap因为我们只想在必要时从数据库中获取（如果你把 2 放在这里，你可能会过度阅读，但会有更少的延迟管道）。

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

于 2016-11-20T10:31:50.953 回答