嗨，我有一个简单的问题，假设我有一个类如下：
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中使用相同的执行器 - 是个好主意？
不要从Scheduler不断重新创建ExecutorService，但要努力将其直接包装在构造函数中。
您根本不需要CompletableFuture，subscribeOn应该应用于flatMap的内部，可能会为每个resize任务挑选单独的线程（它会根据它应用的每个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();
}
所以我的所有过程如下所示：
@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);
}
}
我有一个带图像网址和大小的命令：
@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操作。我应该使用专用调度程序吗？最后我调整了操作，在map操作中创建了请求 - imageResizeRequestFactory。