[设计模式] 模板方法模式与多线程结合实现数据的批处理

一、需求场景

1. 对原始数据进行分析

结果包含三类数据，分别是热点问题、疏整促数据、高频点位。

三种数据结果表均包含日月时间维度，热点问题、疏整促数据除了时间维度还包含全市、区、街道（和村）三个（或者四个）区域维度。

热点问题、疏整促两类数据的结果表字段有很多相似之处。例如某种数据ID和数量、在全市（或区或街乡镇）的占比、当日数量、（市或区或街乡镇）排名、环（同）比量、环（同）比率、环（同）比排名变化、

2. 批处理执行流程

①初始化数据库查询参数

② 从数据取出当日、日累计（指定日之间的累计）、同比累计、环比累计数据

③处理从数据库中获取到的数据

④ 将结果插入结果表

3. 选择技术

① 以热点问题为例，要从数据库取出当日、日累计、同比累计、环比累计四类数据，所以每跑一次要进行四次sql查询，对四个sql的结果进行处理，所以非常适合使用多线程进行异步处理

② 因为每个批处理的执行流程相同，只是每一步的实现不同，同时像“初始化时间参数”、“排名计算”等操作每个批处理的处理相同，所以很适合使用模板方法模式进行设计

二、模板方法模式（参考菜鸟教程）

在模板模式（Template Pattern）中，一个抽象类公开定义了执行它的方法的方式/模板。它的子类可以按需要重写方法实现，但调用将以抽象类中定义的方式进行。这种类型的设计模式属于行为型模式。

意图：定义一个操作中的算法的骨架，而将一些步骤延迟到子类中。模板方法使得子类可以不改变一个算法的结构即可重定义该算法的某些特定步骤。

主要解决：一些方法通用，却在每一个子类都重新写了这一方法。

何时使用：有一些通用的方法。

如何解决：将这些通用算法抽象出来。

关键代码：在抽象类实现，其他步骤在子类实现。

应用实例： 1、在造房子的时候，地基、走线、水管都一样，只有在建筑的后期才有加壁橱加栅栏等差异。 2、西游记里面菩萨定好的 81 难，这就是一个顶层的逻辑骨架。 3、spring 中对 Hibernate 的支持，将一些已经定好的方法封装起来，比如开启事务、获取 Session、关闭 Session 等，程序员不重复写那些已经规范好的代码，直接丢一个实体就可以保存。

优点： 1、封装不变部分，扩展可变部分。 2、提取公共代码，便于维护。 3、行为由父类控制，子类实现。

缺点：每一个不同的实现都需要一个子类来实现，导致类的个数增加，使得系统更加庞大。

使用场景： 1、有多个子类共有的方法，且逻辑相同。 2、重要的、复杂的方法，可以考虑作为模板方法。

注意事项：为防止恶意操作，一般模板方法都加上 final 关键词。

三、多线程技术

以从数据库取出数据为例，每个批处理要从数据库查询四种数据（四个sql），分别是当日、日累计、同比累计、环比累计，对这四个sql开启四个线程，当这四个线程全部从数据库中获取到之后才能继续后面的操作。

1. 如何判断四个线程都已经结束？

这里采用的是线程池，将四个线程放在线程池中判断线程池中的线程是否全部结束（awaitTermination方法），结束代表可以向下进行。其实在这里使用栅栏更合适。

2. 多个全市某天的热点问题批处理同时运行如何保证线程安全？

采用ThreadLocal和InheritableThreadLocal来保证每个线程拥有自己的变量并且子线程可以访问。其实这里可以不在成员变量处对其初始化，同时多个全市某天的热点问题批处理类采用多例实现（@Component注解为单例模式），这样就避免了这个问题。

3. 如何配置线程池参数？

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {

corePoolSize：基本线程数量

maximumPoolSize：最大线程池数量

keepAliveTime：空闲线程最大存活时间

unit：存活时间单位

workQueue：工作阻塞队列

threadFactory：线程工厂

handler：拒绝线程策略

/*** 自定义线程名字*/private final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("ConcurrentHQS-handler-task-pool-%d").build();// 创建用于初始化数据的线程池threadPoolExecutorInitData.set(new ThreadPoolExecutor(3, 5, 1, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(5), threadFactory,new ThreadPoolExecutor.AbortPolicy()));

参考下面的线程池线程数量计算公式：

服务器cpu核心数量为4。

我最终设计corePoolSize为3，maximumPoolSize为5，空闲线程最大存活时间为1分钟，工作阻塞队列采用有界队列5（程序内保证不会超过5）、线程工厂用于记录线程名字便于区分、拒绝线程策略使用AbortPolicy丢弃任务并抛出RejectedExecutionException（程序内保证不会出现）

4. 目前程序的速度瓶颈在于哪里？

随着源数据表数据量的增大，查询sql变得越来越慢，已经优化了索引和sql语句，因为是批处理所以允许查询速度比较慢的情况，毕竟每天只运行一次该sql。

实现代码：

/*** 初始化数据线程池*/private ThreadLocal<ExecutorService> threadPoolExecutorInitData = new ThreadLocal<>();/*** 插入表的数据*/private InheritableThreadLocal<Map<String, DmCityDay>> dmRrpTypeInsertList = new InheritableThreadLocal<>();/*** 同比数据 前一个月中的疏整促数量(返回的是group by 地区,questionId*/private InheritableThreadLocal<Map<String, DmCityDay>> dmRrpTypeM2MList = new InheritableThreadLocal<>();/*** 环比数据 前一年中的疏整促数量(返回的是group by 地区,questionId*/private InheritableThreadLocal<Map<String, DmCityDay>> dmRrpTypeY2YList = new InheritableThreadLocal<>();@Overridepublic void initDataFromSql() {// 创建用于初始化数据的线程池threadPoolExecutorInitData.set(new ThreadPoolExecutor(3, 5, 1, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(5), threadFactory, new ThreadPoolExecutor.AbortPolicy()));// 初始化日累计数据(直接保存到各自对应的Map对象里，下同)Future<Map<String, DmCityDay>> insertData = threadPoolExecutorInitData.get().submit(() -> this.initInsertData(dailyCuRequest.get()));// 初始化同比数Future<Map<String, DmCityDay>> y2YData = threadPoolExecutorInitData.get().submit(() -> this.initY2YData(y2yRequest.get()));// 初始化环比数据Future<Map<String, DmCityDay>> m2MData = threadPoolExecutorInitData.get().submit(() -> this.initM2MData(m2mRequest.get()));// 关闭线程池threadPoolExecutorInitData.get().shutdown();try {// 等待线程结束（超过3小时抛异常）if (!threadPoolExecutorInitData.get().awaitTermination(3, TimeUnit.HOURS)) {log.info("线程全部执行结束");}// 将结果保存起来if (insertData.get() != null) {dmRrpTypeInsertList.set(insertData.get());}if (y2YData.get() != null) {dmRrpTypeY2YList.set(y2YData.get());}if (m2MData.get() != null) {dmRrpTypeM2MList.set(m2MData.get());}} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}}

四、构建思路

抽象基类中定义公共方法，包含：初始化请求参数（initRequest）、sql查询（initDataFromSql），处理日累计、环比、同比数据对象交集数据（handlerJointData）、插入日表数据库（insertDay）、插入月表数据库（insertMonth）、热点问题批处理（调用的方法均为子类去实现）（dmHotQuestion）、疏整促批处理（调用的方法均为子类去实现）（dmRrpType）子类实现抽象基类所未完成的函数调用方使用dmHotQuestion函数和dmRrpType函数调用

五、应用实例

AbstractHandlerTaskTemplate：创建一个抽象类，它的模板方法被设置为 final。

HQCTemplate、HQDTemplate、HQSTemplate：分别是热点问题市、区、街道的模板实现类

RTCTemplate、RTDTemplate、RTSTemplate：分别是疏整促市、区、街道的模板实现类

六、可提高的点

判断四个线程都已经结束采用栅栏（CyclicBarrier）替代目前的新建线程池等待线程池中的线程运行结束不使用Spring注解的单例模式来避免使用ThreadLocal和InheritableThreadLocal，能简化代码和提升代码的可阅读性模板方法中可以把更多的公共方法放在抽象类中，目前是使用一个工具类来多次调用线程池参数缺少调研，只是根据经验估计降低代码时间复杂度，旧代码中的排名计算时间复杂度O（n^2），采用目前新的计算放啊时间复杂度为O（n）

七、经验总结与问题记录

如何实现子线程与主线程共享对象？

InheritableThreadLocal

如何监测线程？

Jprofiler