Flux中并行请求的有序流

Question

我正在寻找一种使用Flux实现以下目标的方法：

1. 处理1,000,000（或更多）请求流。
2. 请求被编号（从1到1,000,000）。
3. 请求应该使用10个线程并行启动。
4. 请求的启动顺序由其序列号决定。
5. 结果通量应该以与请求相同的顺序返回响应。
6. 结果通量应该尽快发出每个响应，因为它的所有前身都是可用的。

我知道#4的答案是为计划程序使用一个执行器。然而，我不确定如何实现#6。

下面是一个示例场景：

>

发出响应3、4、5，并开始对其进行处理。

那么——我应该如何修改下面的代码来实现#6？

public class Example {

private final Scheduler scheduler = Schedulers.fromExecutor(
	Executors.newFixedThreadPool( 10 ) );

public void start() {
	Flux<Request> requestFlux = getFluxOfOneMillionRequests(); // Never mind how this is achieved
	
	Flux<Response> responseFlux = flux.flatMap(request -> doInWorkerThread(request));
	
	flux.doOnNext(response -> processResponse(response)).subscribe()
}

private Mono<Response> doInWorkerThread(Request request) {

	return Mono.fromCallable(() -> {

		// Do something
		return new Response( request.getSerial(), someResult );
	}).subscribeOn(scheduler);
}

private void processResponse(Response response) {
	// Do something
}
}

Answer 1

我想出了一个确定的答案，尽管我求助于使用低级线程同步代码。这个想法是有3个Flux步骤：

• 第一步是在多个线程中处理请求，方法是在调度器上订阅Mono，该调度器使用一个执行器和多个工作线程。当工作线程完成时，它将其响应放在一个公共列表中，并返回一些任意对象，以便激活第二步

对于多线程调度器，我使用一个执行器，以便按照任务插入的顺序执行任务。这将减少缓存机制必须保存的平均响应量。

以下代码片段演示了这一点：

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Executors;

public class StreamSort {

    private final Scheduler produceScheduler = Schedulers.fromExecutor(Executors.newFixedThreadPool(3, new ThreadFactoryBuilder().setNameFormat("--> Producer-%d").build()));
    private final Scheduler consumeScheduler = Schedulers.fromExecutor(Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("<-- Consumer-%d").build()));
    private final ResponseReadySignal responseReadySignal = new ResponseReadySignal();
    private int nextExpected = 1;
    private List<Response> arrivals = Lists.newArrayList();

    private static class ResponseReadySignal {}
    private Random rand = new Random();

    public static void main(String[] args) throws InterruptedException {

        StreamSort m = new StreamSort();

        m.start();
    }

    private void start() throws InterruptedException {

        Flux<Request> requests = generateRequestFlux();

        requests
                .flatMap(request -> doInWorkerThread(request))
                .flatMap(signal -> jumpToCollectingThread(signal))
                .flatMap(signal -> departInOrder(signal))
                .doOnNext(response -> log("Got " + response ))
                .doOnComplete(() -> stopWaiting())
                .subscribe();

        waitUntilFinished();

        log("Disposing schedulers");
        produceScheduler.dispose();
        consumeScheduler.dispose();
        log("FINISH");
    }

    private synchronized void waitUntilFinished() throws InterruptedException {
        this.wait();
    }

    private synchronized void stopWaiting() {
        this.notifyAll();
    }

    private Flux<Request> generateRequestFlux() {
        List<Integer> serials = Lists.newArrayList(1,2,3,4,5,6,7,8,9,10);

        return Flux.fromIterable(serials).map(serial -> new Request(serial));
    }

    private Mono<ResponseReadySignal> doInWorkerThread(Request request) {

        return Mono.fromCallable(() -> {

            Response response = handleRequest(request);

            synchronized (this) {
                arrivals.add(response);
            }

            return responseReadySignal;
        }).subscribeOn(produceScheduler);
    }

    private Response handleRequest(Request request) throws InterruptedException {
        int milli = rand.nextInt(1000);

        log( request + " start, sleeping " + milli );

        Thread.currentThread().sleep( milli ); // Simulate some task that takes time

        log( request + " Finished" );

        return new Response(request.getSerial()); // Simulate response creation.

    }

    private Mono<ResponseReadySignal> jumpToCollectingThread(ResponseReadySignal signal) {

        log( "Delivering signal to collecting thread" );

        return Mono.fromCallable(() -> signal).subscribeOn(consumeScheduler);
    }

    private Flux<Response> departInOrder(ResponseReadySignal signal) {

        List<Response> readyToDepart = Lists.newLinkedList();

        synchronized (this) {
            Collections.sort(arrivals);
            while (arrivals.size() > 0 && arrivals.get(0).getSerial() == nextExpected) {
                readyToDepart.add(arrivals.remove(0));
                ++nextExpected;
            }
        }

        log("Departing " + readyToDepart.size() + " items: " + readyToDepart);
        return Flux.fromIterable(readyToDepart);
    }

    private static void log( String message ) {
        Thread t = Thread.currentThread();
        System.out.println( t.getName() + ": " + message );

    }

    private static class WithSerial implements Comparable<WithSerial> {
        private final int serial;

        public WithSerial(int serial) {
            this.serial = serial;
        }

        public int getSerial() {
            return serial;
        }

        @Override
        public int compareTo(WithSerial o) {
            return this.serial - o.serial;
        }

        @Override
        public String toString() {
            return "" + getSerial();
        }
    }

    private static class Request extends WithSerial {

        public Request(int serial) {
            super(serial);
        }
    }

    private static class Response extends WithSerial {
        public Response(int serial) {
            super(serial);
        }
    }
}