我可以为 subscribeOn 方法和异步任务使用相同的执行程序吗

Question

您好，我有一个简单的问题，假设我有一个像下面这样的课程：

import lombok.Value;

import java.nio.file.Path;

@Value
class ImageResizeRequest {

    private DownloadedImage downloadedImage;

    private ImageSize imageSize;

    private Path destinationLocation;
}

上面的类表示负责将图像大小调整为给定大小的单个任务。我有很多要求将此图像调整为许多不同的尺寸。

@RequiredArgsConstructor
class ImageResizeService {

    private final Executor executor;

    Mono<List<ImageResizeResult>> resize(List<ImageResizeRequest> requests) {

        return Flux.fromIterable(requests)
                .flatMap(this::resize)
                .collectList()
                .subscribeOn(Schedulers.fromExecutor(executor));
    }

    private Mono<ImageResizeResult> resize(ImageResizeRequest request) {

        return Mono.fromFuture(CompletableFuture.supplyAsync(resizeTask(request), executor));

    }

    private Supplier<ImageResizeResult> resizeTask(ImageResizeRequest request) {
        return () -> {
            //TODO add image resize logic for example ImageMagick by Im4Java...
            /** code below call ImageMagick library
             ConvertCmd cmd = new ConvertCmd();
             IMOperation op = new IMOperation();
             op.quality(100d);
             op.addImage(request.getDestinationLocation().toString());
             cmd.run(op);

             */
            //TODO add logic!!!
            return new ImageResizeResult(null, null, null, null);
        };
    }
}

我的问题是： 如何在 Project Reactor 中实现负责调整图像大小的并行独立任务？如果没有 Project reactor，我将使用 CompletableFuture 列表：

private static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> futures) {
    CompletableFuture<Void> allDoneFuture =
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
    return allDoneFuture.thenApply(v ->
            futures.stream().
                    map(future -> future.join()).
                    collect(Collectors.<T>toList())
    );
}

具有指定的执行器服务。此外，在我的示例中，我在 subscribeOn 方法和 supplyAsync 中使用相同的执行程序 - 是个好主意吗？

Answer 1

不要不断地从ExecutorService 重新创建Scheduler，而是努力将其直接包装在构造函数中。

您根本不需要CompletableFuture，并且应该将subscribeOn 应用到flatMap 的内部，以便为每个调整大小任务选择单独的线程（它会从每个Flux 池中选择一个线程）适用于）：

class ImageResizeService {

  private final Executor executor; //TODO prefer an ExecutorService if possible
  private final Scheduler scheduler; //FIXME Schedulers.fromExecutor(executor)

  Mono<List<ImageResizeResult>> resize(List<ImageResizeRequest> requests) {
    //we get the requests on IO thread
    return Flux.fromIterable(requests)
            //for each request, perform asynchronous resize...
            .flatMap(r -> Mono
                //... by converting the resizeTask Callable to a Mono
                .fromCallable(r -> resizeTask(r).get())
                //... and making sure it executes on the executor
                .subscribeOn(scheduler)
            )
            .collectList();
  }
}

为了实现真正的并行化，您还有另一个选择：parallel().runOn():

Mono<List<ImageResizeResult>> resize(List<ImageResizeRequest> requests) {
    //we get the requests on IO thread
    return Flux.fromIterable(requests)
            //divide into N workloads
            //the executor _should_ be capable of this degree of parallelisation:
            .parallel(NUMBER_OF_DESIRED_THREADS)
            //actually tell to run each workload on a thread picked from executor
            .runOn(scheduler) 
            //here the workload are already running on their dedicated thread,
            //we can afford to block it and thus apply resize in a simpler `map`
            .map(r -> resizeTask(r).get()) //NB: the Supplier aspect can probably be removed
            //go back to a `Flux` sequence for collection into list
            .sequential()
            .collectList();
}

Answer 2

所以我的所有过程如下所示：

@RequiredArgsConstructor
class ImageCommandProcessingService {

    private final DownloadRequestFactory downloadRequestFactory;
    private final ImageClientDownloader imageClientDownloader;
    private final ImageResizeRequestFactory imageResizeRequestFactory;
    private final ImageResizeService imageResizeService;

    Mono<List<ImageResizeResult>> process(ResizeImageCommand resizeImageCommand) {
        return Mono.just(resizeImageCommand)
                .map(command -> downloadRequestFactory.create(command.getImageUrl().getUrl()))
                .flatMap(imageClientDownloader::downloadImage)
                .map(downloadedImage -> imageResizeRequestFactory.createRequests(downloadedImage, resizeImageCommand.getSizes().toJavaList()))
                .flatMap(imageResizeService::resize);

    }

}

我有一个带有图像 url 和一组尺寸的命令：

@Value
class ResizeImageCommand {

    private ImageUrl imageUrl;

    private Set<ImageSize> sizes;
}

首先我需要在磁盘上下载图像，所以我通过工厂创建一个下载请求：

@RequiredArgsConstructor
class DownloadRequestFactory {

    private final ImageLocationPathResolver resolver;

    DownloadRequest create(String url) {
        return new DownloadRequest(url, resolver.resolveDownloadedLocation(url));
    }
}

Resolver 是一个类，负责创建临时文件的路径，并为调整大小的图像创建路径：

class ImageLocationPathResolver {

    private String temporaryImagesFolder;
    private String destinationImagesFolder;

    Path resolveDownloadedLocation(String imageUrl) {
        LocalDateTime now = LocalDateTime.now();
        String fileName = now.toString() + "_" + getFileNameExtensionFromUrl(imageUrl);
        return Paths.get(temporaryImagesFolder,getDatePaths(now.toLocalDate()), fileName);
    }

    Path resolveDestinationLocation(ImageSize imageSize, String url) {
        String fileName = getFileNameExtensionFromUrl(url);
        return Paths.get(destinationImagesFolder, imageSize.getName(), getDatePaths(LocalDate.now()), fileName);
    }

    private String getFileNameExtensionFromUrl(String url) {
        return StringUtils.getFilenameExtension(url);
    }

    private String getDatePaths(LocalDate now) {
        return now.getYear() + File.pathSeparator + now.getMonth() + File.pathSeparator + now.getDayOfMonth();
    }
}

我还有一个负责下载操作的客户：

public interface ImageClientDownloader {

    Mono<DownloadedImage> downloadImage(DownloadRequest downloadRequest);
}

和实施：

@Slf4j
class HttpImageClientDownloader implements ImageClientDownloader {

    private final WebClient webClient;

    HttpImageClientDownloader() {
        this.webClient = WebClient.create();
    }

    @Override
    public Mono<DownloadedImage> downloadImage(DownloadRequest downloadRequest) {
        try {
            Flux<DataBuffer> dataBuffer = webClient.get()
                    .uri(downloadRequest.getUrl())
                    .retrieve()
                    .bodyToFlux(DataBuffer.class);


            Path resultFilePath = Files.createFile(downloadRequest.getLocation());
            WritableByteChannel channel = Files.newByteChannel(resultFilePath, StandardOpenOption.WRITE);
            return DataBufferUtils.write(dataBuffer, channel)
                    .map(DataBufferUtils::release)
                    .then(Mono.just(new DownloadedImage(downloadRequest.getUrl(), resultFilePath, LocalDateTime.now())));

        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return Mono.error(e);
        }
    }
}

是IO操作。我应该使用专用调度程序吗？ 最后我进行了调整大小操作，在地图操作中创建了请求 - imageResizeRequestFactory。