【Java设计模式-源码】扇出扇入模式

@[TOC] # 【Java设计模式】扇出扇入模式 ## 一、概述 Java中的扇出扇入设计模式旨在通过将一个任务划分为多个可以并行处理的子任务（扇出），然后将这些子任务的结果合并为一个单一的结果（扇入），来提高并发性和优化处理时间。 ## 二、别名 * 分散收集 ## 三、扇出扇入设计模式的意图 扇出扇入设计模式的目的是在Java中通过并行处理多个子任务并合并结果，来提高并发性能和优化处理效率。 ## 四、通过实际示例详细解释扇出扇入模式 实际示例： > Java中扇出扇入模式的一个现实世界例子是像UberEats或DoorDash这样的食品配送服务。当顾客下单时，服务（扇出）向不同的餐厅发送单独的任务来准备各种菜品。每个餐厅独立工作来准备订单的一部分。一旦所有餐厅完成任务，配送服务（扇入）将来自不同餐厅的菜品聚合为一个订单，确保所有东西一起交付给顾客。这种并行处理提高了效率并确保及时交付。 通俗解释： > 扇出扇入模式将任务分布在多个并发进程或线程中，然后聚合结果。 维基百科说： > 在面向消息的中间件中，扇出模式通过并行将消息传递到一个或多个目的地，而不等待响应，来模拟信息交换。这允许一个进程同时将任务分配给各种接收器。 > > 另一方面，扇入概念通常指的是多个输入的聚合。在数字电子学中，它描述了一个逻辑门可以处理的输入数量。结合这些概念，软件工程中的扇出扇入模式涉及分发任务（扇出），然后聚合结果（扇入）。 ## 五、Java中扇出扇入模式的编程示例 所提供的实现涉及一个数字列表，目标是对它们进行平方并聚合结果。`FanOutFanIn`类接收数字列表作为`SquareNumberRequest`对象和一个`Consumer`实例，该实例在请求完成时收集平方后的结果。每个`SquareNumberRequest`以随机延迟对其数字进行平方，并在平方后调用`Consumer`。`Consumer`实例在不同时间从各个`SquareNumberRequest`对象获取结果。 以下是Java中的`FanOutFanIn`类，它通过异步分发请求来演示扇出扇入模式。 ```java public class FanOutFanIn { public static Long fanOutFanIn(final List<SquareNumberRequest> requests, final Consumer consumer) { ExecutorService service = Executors.newFixedThreadPool(requests.size()); // 扇出 List<CompletableFuture<Void>> futures = requests .stream() .map(request -> CompletableFuture.runAsync(() -> request.delayedSquaring(consumer), service)) .collect(Collectors.toList()); CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); return consumer.getSumOfSquaredNumbers().get(); } } ``` `Consumer`是一个回调类，当请求完成时将被调用。它将聚合所有请求的结果。 ```java public class Consumer { private final AtomicLong sumOfSquaredNumbers; Consumer(Long init) { sumOfSquaredNumbers = new AtomicLong(init); } public Long add(final Long num) { return sumOfSquaredNumbers.addAndGet(num); } } ``` 请求表示为`SquareNumberRequest`，它以随机延迟对数字进行平方，并在平方后调用`Consumer`。 ```java public class SquareNumberRequest { private final Long number; public void delayedSquaring(final Consumer consumer) { var minTimeOut = 5000L; SecureRandom secureRandom = new SecureRandom(); var randomTimeOut = secureRandom.nextInt(2000); try { // 这将使线程睡眠5 - 7秒。 Thread.sleep(minTimeOut + randomTimeOut); } catch (InterruptedException e) { LOGGER.error("睡眠时发生异常", e); Thread.currentThread().interrupt(); } finally { consumer.add(number * number); } } } ``` 以下是带有主方法的`App`类，用于驱动示例。 ```java public static void main(String[] args) { final List<Long> numbers = Arrays.asList(1L, 3L, 4L, 7L, 8L); LOGGER.info("要平方并求和的数字 --> {}", numbers); final List<SquareNumberRequest> requests = numbers.stream().map(SquareNumberRequest::new).toList(); var consumer = new Consumer(0L); // 将请求和消费者传递给fanOutFanIn，有时也称为协调器函数 final Long sumOfSquaredNumbers = FanOutFanIn.fanOutFanIn(requests, consumer); LOGGER.info("所有平方数的和 --> {}", sumOfSquaredNumbers); } ``` 运行示例将产生以下控制台输出。 ``` 06:52:04.622 [main] INFO com.iluwatar.fanout.fanin.App -- 要平方并求和的数字 --> [1, 3, 4, 7, 8] 06:52:11.465 [main] INFO com.iluwatar.fanout.fanin.App -- 所有平方数的和 --> 139 ``` ## 六、何时在Java中使用扇出扇入模式 Java中的扇出扇入设计模式适用于任务可以分解并并行执行的场景，特别适合于数据处理、批处理以及需要从各种来源聚合结果的情况。 ## 七、扇出扇入模式在Java中的实际应用 * Java中的扇出扇入模式广泛应用于大规模数据处理应用程序。 * 在需要从多个来源聚合结果后再提供响应的服务中，例如在分布式缓存或负载均衡系统中。 ## 八、扇出扇入模式的好处和权衡 好处： * 通过并行处理提高性能。 * 提高系统的响应能力。 * 有效利用多核处理器架构。 权衡： * 错误处理的复杂性增加。 * 由于任务同步和结果聚合，可能会增加开销。 * 依赖于底层基础设施支持并发执行的能力。 ## 九、源码下载 [扇出扇入模式示例代码下载](链接待补充)