java - 按优先级调度周期性线程

Question

我必须在 java 8 中实现一种多线程应用程序。

我的应用程序应该定期运行一些线程，假设每 30 分钟运行一次。每个线程都应该调用一个 api，从中获取数据并将其保存到数据库中。数据库的组织方式如下:

Table A
Table B
Table C
Table D
Table E
Table F
Table G
Table H

Tables F , G和H有外键 Tables A , B , C , D和E .

这导致有一个 Thread对于每个要更新的表:

Thread A -> Table A
Thread B -> Table B
Thread C -> Table C
Thread D -> Table D
Thread E -> Table E
Thread F -> Table F
Thread G -> Table G
Thread H -> Table H

由于某些表具有其他表的外键，因此线程不应同时启动，而应首先启动 Thread A , B , C和E一旦完成，就应该开始 Thread F , G和H .

我在互联网上搜索了很多可能的解决方案，但没有找到适合我情况的解决方案。

目前我已经用 ScheduledThreadPoolExecutor 实现了我的代码其功能为 scheduleWithFixedDelay这确保了线程每 30 分钟执行一次。即使我以不同的优先级安排每个线程，它们也会同时启动，并且在使用外键的表中执行插入/更新时会出现异常。

这是我的代码的快照，可能有助于理解。

Thread-like类看起来像这样:

public class AllarmReasonService implements Runnable {

    @Override
    public void run() {
        AllarmReasonDAO dao = new AllarmReasonDAO();
        PagedResultContainer<AllarmReason> allarmReasonContainer = AllarmReasonApi.getInstance().getAllarmReasons();
        for(Iterator<AllarmReason> iterator = allarmReasonContainer.getData().iterator(); iterator.hasNext();){
            AllarmReason allarmReason = iterator.next();
            dao.insertOrUpdateAllarmReason(allarmReason);
        }
    }
}

threadExecutor :

public void threadExecutor(){
    initializeThreadMap();
    Stream<Entry<String, Integer>> sorted = threadTimerMap.entrySet().stream().sorted(Entry.comparingByValue());
    scheduler = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(threadTimerMap.size());

    for(Iterator<Entry<String, Integer>> entryIterator = sorted.iterator(); entryIterator.hasNext();){
        Entry<String, Integer> entry = entryIterator.next();
        Runnable service = threadMap.get(entry.getKey());
        Thread thread = new Thread(service);
        thread.setPriority(entry.getValue());
        scheduler.scheduleWithFixedDelay(thread, 0, 30 * 60 * 1000, TimeUnit.MILLISECONDS);
    }
}

private void initializeThreadMap(){
        threadMap.clear();
        threadMap.put("causaliAllarme", new AllarmReasonService());
        threadMap.put("causaliLavorazione", new ManufactureReasonService());
        threadMap.put("celle", new CellService());
        threadMap.put("dipendenti", new EmployeeService());
        threadMap.put("ordiniLavorazione", new ManufactureOrderService());
        threadMap.put("parzialiLavorazione", new ManufacturePartialService());
        threadMap.put("qualita", new QualityService());
        threadMap.put("qualitaRilevata", new QualityDetectionService());
    }

最后，threadTimerMap有key = name of the module/thread (i.e. causaliAllarme)和value = priority其中首先更新模块的优先级较高。

通过此设置，即使我设置了一定的优先级，我也无法首先运行最重要的线程。

考虑到我不是线程和多线程方面的专家。

任何帮助将不胜感激。提前致谢。

Answer 1

对于这种情况，我建议使用 CompletableFuture特点。

// Declare your pool. Declare ThreadFactory also, it will be very
// helpful with debug, I promise :) 
int corePoolSize = 10;
ScheduledExecutorService pool = Executors.newScheduledThreadPool(corePoolSize,
      new ThreadFactoryBuilder().setNameFormat("Your-thread-%d").setDaemon(true).build());

List<CompletableFuture<Void>> dependencies = new ArrayList<>();

// Submit the threads for the first stage
dependencies.add(CompletableFuture.runAsync(new AllarmReasonService(), pool));
dependencies.add(CompletableFuture.runAsync(new ManufactureReasonService(), pool));
// ...
// do the same with all your stage-1 threads

// wait while stage 1 completed
try {
    for (CompletableFuture<Void> f : dependencies) {
        f.get();
    }
} catch (InterruptedException | ExecutionException e) {
    // log or re-throw
    pool.shutdownNow();
}

// stage 2
CompletableFuture.runAsync(new AllarmReasonService(), pool);
CompletableFuture.runAsync(new ManufactureReasonService(), pool);
// other required ...

您还可以使用 CompletableFuture.allOf(CompletableFuture<?>...) 聚合 future 方法并等待单个 future 。

希望对你有帮助!