江苏龙网

  1. 主页 > 生活百科 > >

DDD死党：内存Join——将复用和扩展用到极致( 六 )

DDD

终于，我们将变化收敛到 VO 内 ， VO 需要绑定新的数据，只需实现对应接口即可 。
4.3. 并发绑定经过重构，代码结构变得非常清晰，如果想通过多线程并发方式提供性能，需要调整哪些组件呢？好好想想?。。?
只需对FetcherService进行调整 ， 让我们来一个并发版本，具体如下：
@Servicepublic class ConcurrentFetcherService {private ExecutorService executorService;@Autowiredprivate List<ItemFetcherExecutor> itemFetcherExecutors;@PostConstructpublic void init(){this.executorService = Executors.newFixedThreadPool(20);}@SneakyThrowspublic <F extends ItemFetcher> void fetch(Class<F> cls, List<F> fetchers){if (CollectionUtils.isNotEmpty(fetchers)){// 创建异步执行任务List<Callable<Void>> callables = this.itemFetcherExecutors.stream().filter(itemFetcherExecutor -> itemFetcherExecutor.support(cls)).map(itemFetcherExecutor -> (Callable<Void>) () -> {itemFetcherExecutor.fetch(fetchers);return null;}).collect(Collectors.toList());// 线程池中并行执行this.executorService.invokeAll(callables);}}}OrderServiceFetcherV3 只需使用 ConcurrentFetcherService 替代 原来的 FetcherService 并拥有了并发能力 。
5. 注解方案5.1. 复杂配置 @JoinInMemory 来帮忙纵观整个 Fetcher 封装，虽然结构清晰 ， 但细节过于繁琐，特别是：

  1. 待抓取数据需要抽取 Fetcher 接口
  2. 需要提供自己的 FetcherExecutor 实现
  3. VO 需要实现多个 Fetcher 接口
这些不便将成为落地最大的阻碍，那有没有办法进行进一步简化？
这需要思考下这些设计背后的深层需求：
Fetcher接口目的包括
  1. 提供绑定信息
  2. 设置绑定结果
  3. 被 FetcherExecutor 识别并进行处理
FetcherExecutor设计的目标包括：
  1. 识别待处理的 Fetcher
  2. 定制个性化流程
所有这些需求是否可用 ==注解== 的方式实现？
  1. 在 VO 属性上增加注解，说明绑定结果写回到该属性上
  2. 注解配置来源属性，提供绑定信息
  3. 注解配置流程属性，完成 FetcherExecutor 的个性化定制
  4. 每个注解背后是一个 FetcherExecutor 实现，完成 FetcherExecutor 与 “Fetcher” 绑定
根据上述分析 ， 注解可完成全部任务，新建注解如下：
@Target({ElementType.FIELD, ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)public @interface JoinInMemory {/*** 从 sourceData 中提取 key* @return*/String keyFromSourceData();/*** 从 joinData 中提取 key* @return*/String keyFromJoinData();/*** 批量数据抓取* @return*/String loader();/*** 结果转换器* @return*/String joinDataConverter() default "";/*** 运行级别，同一级别的 join 可 并行执行* @return*/int runLevel() default 10;}乍一看 ， 需要配置的信息真多，其实大多数配置全部与 FetcherExecutor 实现相关 。
abstract class AbstractJoinItemExecutor<SOURCE_DATA, JOIN_KEY, JOIN_DATA, JOIN_RESULT> implements JoinItemExecutor<SOURCE_DATA> {/*** 从原始数据中生成 JoinKey* @param data* @return*/protected abstract JOIN_KEY createJoinKeyFromSourceData(SOURCE_DATA data);/*** 根据 JoinKey 批量获取 JoinData* @param joinKeys* @return*/protected abstract List<JOIN_DATA> getJoinDataByJoinKeys(List<JOIN_KEY> joinKeys);/*** 从 JoinData 中获取 JoinKey* @param joinData* @return*/protected abstract JOIN_KEY createJoinKeyFromJoinData(JOIN_DATA joinData);/*** 将 JoinData 转换为 JoinResult* @param joinData* @return*/protected abstract JOIN_RESULT convertToResult(JOIN_DATA joinData);/*** 将 JoinResult 写回至 SourceData* @param data* @param JoinResults*/protected abstract void onFound(SOURCE_DATA data, List<JOIN_RESULT> JoinResults);/*** 未找到对应的 JoinData* @param data* @param joinKey*/protected abstract void onNotFound(SOURCE_DATA data, JOIN_KEY joinKey);@Overridepublic void execute(List<SOURCE_DATA> sourceDatas) {// 从源数据中提取 JoinKeyList<JOIN_KEY> joinKeys = sourceDatas.stream().filter(Objects::nonNull).map(this::createJoinKeyFromSourceData).filter(Objects::nonNull).distinct().collect(toList());log.debug("get join key {} from source data {}", joinKeys, sourceDatas);// 根据 JoinKey 获取 JoinDataList<JOIN_DATA> allJoinDatas = getJoinDataByJoinKeys(joinKeys);log.debug("get join data {} by join key {}", allJoinDatas, joinKeys);// 将 JoinData 以 Map 形式进行组织Map<JOIN_KEY, List<JOIN_DATA>> joinDataMap = allJoinDatas.stream().filter(Objects::nonNull).collect(groupingBy(this::createJoinKeyFromJoinData));log.debug("group by join key, result is {}", joinDataMap);// 处理每一条 SourceDatafor (SOURCE_DATA data : sourceDatas){// 从 SourceData 中 获取 JoinKeyJOIN_KEY joinKey = createJoinKeyFromSourceData(data);if (joinKey == null){log.warn("join key from join data {} is null", data);continue;}// 根据 JoinKey 获取 JoinDataList<JOIN_DATA> joinDatasByKey = joinDataMap.get(joinKey);if (CollectionUtils.isNotEmpty(joinDatasByKey)){// 获取到 JoinData，转换为 JoinResult，进行数据写回List<JOIN_RESULT> joinResults = joinDatasByKey.stream().filter(Objects::nonNull).map(joinData -> convertToResult(joinData)).collect(toList());log.debug("success to convert join data {} to join result {}", joinDatasByKey, joinResults);onFound(data, joinResults);log.debug("success to write join result {} to source data {}", joinResults, data);}else {log.warn("join data lost by join key {} for source data {}", joinKey, data);// 为获取到 JoinData，进行 notFound 回调onNotFound(data, joinKey);}}}}


推荐阅读


上一篇：掌握C++模板的艺术：类型参数、默认值和自动推导

下一篇：对 Bash 感到厌倦？教你如何在 Linux 中更改默认 Shell