乐趣区

关于多线程:多线程永动任务设计与实现

jiezi

明天教大家撸一个 Java 的多线程永动工作，这个示例的原型是公司自研的多线程异步工作我的项目，我把外面波及到多线程的代码抽离进去，而后进行肯定的革新。

外面波及的知识点十分多，特地适宜有 肯定工作教训 的同学学习，或者能够间接拿到我的项目中应用。

文章构造非常简单：

1. 性能阐明

做这个多线程异步工作，次要是因为咱们有很多永动的异步工作，什么是永动呢？就是 工作跑起来后，须要始终跑下去。

比方音讯 Push 工作，因为始终有音讯过去，所以须要始终去生产 DB 中的未推送音讯，就须要整一个 Push 的永动异步工作。

咱们的需要其实不难，简略总结一下：

  1. 能同时执行多个永动的异步工作；
  2. 每个异步工作，反对开 多个线程 去生产这个工作的数据；
  3. 反对永动异步工作的 优雅敞开，即敞开后，须要把所有的数据生产结束后，再敞开。

实现下面的需要，须要留神几个点：

  1. 每个 永动工作，能够开一个线程去执行；
  2. 每个 子工作，因为须要反对并发，须要用线程池管制；
  3. 永动工作的敞开，须要告诉子工作的并发线程，并 反对永动工作和并发子工作的优雅敞开。

2. 多线程工作示例

2.1 线程池

对于子工作，须要反对并发，如果每个并发都开一个线程，用完就敞开，对资源耗费太大，所以引入线程池：

public class TaskProcessUtil {
    // 每个工作，都有本人独自的线程池
    private static Map<String, ExecutorService> executors = new ConcurrentHashMap<>();

    // 初始化一个线程池
    private static ExecutorService init(String poolName, int poolSize) {
        return new ThreadPoolExecutor(poolSize, poolSize,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new ThreadFactoryBuilder().setNameFormat("Pool-" + poolName).setDaemon(false).build(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    // 获取线程池
    public static ExecutorService getOrInitExecutors(String poolName,int poolSize) {ExecutorService executorService = executors.get(poolName);
        if (null == executorService) {synchronized (TaskProcessUtil.class) {executorService = executors.get(poolName);
                if (null == executorService) {executorService = init(poolName, poolSize);
                    executors.put(poolName, executorService);
                }
            }
        }
        return executorService;
    }

    // 回收线程资源
    public static void releaseExecutors(String poolName) {ExecutorService executorService = executors.remove(poolName);
        if (executorService != null) {executorService.shutdown();
        }
    }
}

这是一个线程池的工具类，这里初始化线程池和回收线程资源很简略，咱们次要探讨获取线程池。

获取线程池可能会存在并发状况，所以须要加一个 synchronized 锁，而后锁住后，须要对 executorService 进行二次判空校验。

2.2 单个工作

为了更好解说单个工作的实现形式，咱们的工作次要就是把 Cat 的数据打印进去，Cat 定义如下：

@Data
@Service
public class Cat {
    private String catName;
    public Cat setCatName(String name) {
        this.catName = name;
        return this;
    }
}

单个工作次要包含以下性能：

  • 获取永动工作数据：这里个别都是扫描 DB，我间接就简略用 queryData() 代替。
  • 多线程执行工作：须要把数据拆分成 4 份，而后别离由多线程并发执行，这里能够通过线程池反对；
  • 永动工作优雅停机：当里面告诉工作须要停机，须要执行完残余工作数据，并回收线程资源，退出工作；
  • 永动执行：如果未收到停机命令，工作须要始终执行上来。

间接看代码：

public class ChildTask {

    private final int POOL_SIZE = 3; // 线程池大小
    private final int SPLIT_SIZE = 4; // 数据拆分大小
    private String taskName;

    // 接管 jvm 敞开信号，实现优雅停机
    protected volatile boolean terminal = false;

    public ChildTask(String taskName) {this.taskName = taskName;}

    // 程序执行入口
    public void doExecute() {
        int i = 0;
        while(true) {System.out.println(taskName + ":Cycle-" + i + "-Begin");
            // 获取数据
            List<Cat> datas = queryData();
            // 解决数据
            taskExecute(datas);
            System.out.println(taskName + ":Cycle-" + i + "-End");
            if (terminal) {
                // 只有利用敞开，才会走到这里，用于实现优雅的下线
                break;
            }
            i++;
        }
        // 回收线程池资源
        TaskProcessUtil.releaseExecutors(taskName);
    }

    // 优雅停机
    public void terminal() {
        // 关机
        terminal = true;
        System.out.println(taskName + "shut down");
    }

    // 解决数据
    private void doProcessData(List<Cat> datas, CountDownLatch latch) {
        try {for (Cat cat : datas) {System.out.println(taskName + ":" + cat.toString() + ",ThreadName:" + Thread.currentThread().getName());
                Thread.sleep(1000L);
            }
        } catch (Exception e) {System.out.println(e.getStackTrace());
        } finally {if (latch != null) {latch.countDown();
            }
        }
    }

    // 解决单个工作数据
    private void taskExecute(List<Cat> sourceDatas) {if (CollectionUtils.isEmpty(sourceDatas)) {return;}
        // 将数据拆成 4 份
        List<List<Cat>> splitDatas = Lists.partition(sourceDatas, SPLIT_SIZE);
        final CountDownLatch latch = new CountDownLatch(splitDatas.size());

        // 并发解决拆分的数据，共用一个线程池
        for (final List<Cat> datas : splitDatas) {ExecutorService executorService = TaskProcessUtil.getOrInitExecutors(taskName, POOL_SIZE);
            executorService.submit(new Runnable() {
                @Override
                public void run() {doProcessData(datas, latch);
                }
            });
        }

        try {latch.await();
        } catch (Exception e) {System.out.println(e.getStackTrace());
        }
    }

    // 获取永动工作数据
    private List<Cat> queryData() {List<Cat> datas = new ArrayList<>();
        for (int i = 0; i < 5; i ++) {datas.add(new Cat().setCatName("罗小黑" + i));
        }
        return datas;
    }
}

简略解释一下：

  • queryData：用于获取数据，理论利用中其实是须要把 queryData 定为形象办法，而后由各个工作实现本人的办法。
  • doProcessData：数据处理逻辑，理论利用中其实是须要把 doProcessData 定为形象办法，而后由各个工作实现本人的办法。
  • taskExecute：将数据拆分成 4 份，获取该工作的线程池，并交给线程池并发执行，而后通过 latch.await() 阻塞。当这 4 份数据都执行胜利后，阻塞完结，该办法才返回。
  • terminal：仅用于承受停机命令，这里该变量定义为 volatile，所以多线程内存可见；
  • doExecute：程序执行入口，封装了每个工作执行的流程，当 terminal=true 时，先执行完工作数据，而后回收线程池，最初退出。

2.3 工作入口

间接上代码：

public class LoopTask {
    private List<ChildTask> childTasks;
    public void initLoopTask() {childTasks = new ArrayList();
        childTasks.add(new ChildTask("childTask1"));
        childTasks.add(new ChildTask("childTask2"));
        for (final ChildTask childTask : childTasks) {new Thread(new Runnable() {
                @Override
                public void run() {childTask.doExecute();
                }
            }).start();}
    }
    public void shutdownLoopTask() {if (!CollectionUtils.isEmpty(childTasks)) {for (ChildTask childTask : childTasks) {childTask.terminal();
            }
        }
    }
    public static void main(String args[]) throws Exception{LoopTask loopTask = new LoopTask();
        loopTask.initLoopTask();
        Thread.sleep(5000L);
        loopTask.shutdownLoopTask();}
}

每个工作都开一个独自的 Thread，这里我初始化了 2 个永动工作，别离为 childTask1 和 childTask2，而后别离执行，前面 Sleep 了 5 秒后，再敞开工作，咱们能够看看是否能够依照咱们的预期优雅退出。

2.4 后果剖析

执行后果如下：

childTask1:Cycle-0-Begin
childTask2:Cycle-0-Begin
childTask1:Cat(catName= 罗小黑 0),ThreadName:Pool-childTask1
childTask1:Cat(catName= 罗小黑 4),ThreadName:Pool-childTask1
childTask2:Cat(catName= 罗小黑 4),ThreadName:Pool-childTask2
childTask2:Cat(catName= 罗小黑 0),ThreadName:Pool-childTask2
childTask1:Cat(catName= 罗小黑 1),ThreadName:Pool-childTask1
childTask2:Cat(catName= 罗小黑 1),ThreadName:Pool-childTask2
childTask2:Cat(catName= 罗小黑 2),ThreadName:Pool-childTask2
childTask1:Cat(catName= 罗小黑 2),ThreadName:Pool-childTask1
childTask2:Cat(catName= 罗小黑 3),ThreadName:Pool-childTask2
childTask1:Cat(catName= 罗小黑 3),ThreadName:Pool-childTask1
childTask2:Cycle-0-End
childTask2:Cycle-1-Begin
childTask1:Cycle-0-End
childTask1:Cycle-1-Begin
childTask2:Cat(catName= 罗小黑 0),ThreadName:Pool-childTask2
childTask2:Cat(catName= 罗小黑 4),ThreadName:Pool-childTask2
childTask1:Cat(catName= 罗小黑 4),ThreadName:Pool-childTask1
childTask1:Cat(catName= 罗小黑 0),ThreadName:Pool-childTask1
childTask1 shut down
childTask2 shut down
childTask2:Cat(catName= 罗小黑 1),ThreadName:Pool-childTask2
childTask1:Cat(catName= 罗小黑 1),ThreadName:Pool-childTask1
childTask1:Cat(catName= 罗小黑 2),ThreadName:Pool-childTask1
childTask2:Cat(catName= 罗小黑 2),ThreadName:Pool-childTask2
childTask1:Cat(catName= 罗小黑 3),ThreadName:Pool-childTask1
childTask2:Cat(catName= 罗小黑 3),ThreadName:Pool-childTask2
childTask1:Cycle-1-End
childTask2:Cycle-1-End

输入数据：

  • “Pool-childTask”是线程池名称；
  • “childTask”是工作名称；
  • “Cat(catName= 罗小黑)”是执行的后果；
  • “childTask shut down”是敞开标记；
  • “childTask:Cycle-X-Begin”和“childTask:Cycle-X-End”是每一轮循环的开始和完结标记。

咱们剖析一下执行后果：

  • childTask1 和 childTask2 别离执行，在第一轮循环中都失常输入了 5 条罗小黑数据；
  • 第二轮执行过程中，我启动了敞开指令，这次第二轮执行没有间接进行，而是先执行完工作中的数据，再执行退出，所以完全符合咱们的优雅退出论断。

2.5 源码地址

GitHub 地址：

https://github.com/lml200701158/java-study/tree/master/src/main/java/com/java/parallel/pool/ofc
退出移动版