如何使用自己的线程池和其他配置独立运行两个进程？

Question

我有两个processes，如下所示。

public static final ImmutableList<String> processes = ImmutableList.of("processA", "processB");

现在对于每个process，我都有一个不同的Properties 对象，这两个进程之间根本没有关系和依赖关系。它们相互独立。

我编写了一个代码，目前只适用于一个进程，我需要有效地扩展我的设计，以便它可以适用于两个进程。每个进程都应该有自己的线程池配置。例如，我可能想用三个线程运行processA，用两个线程运行processB。

public class ProcessA implements Runnable {
  private final Properties props;
  private final String processName;

  public ProcessA(String processName, Properties props) {
    this.processName = processName;
    this.props = props;
  }

  @Override
  public void run() {
    List<String> clients = getClients(processName);
    try {
      // .. some code here which does  processing
      // calling some classes here as well

    } catch (Exception ex) {
      // log error
    } finally {
      // close processA here
    }
  }

  public void shutdown() {
    // shutdown processA here
  }
}

下面是我执行processA 的主要课程。一般来说，我将只从下面的类执行我的两个过程。

@Singleton
@DependencyInjectionInitializer
public class Initializer {
  private final ExecutorService executorServiceProcessA = Executors.newFixedThreadPool(3);
  private final List<ProcessA> processAList = new ArrayList<>();

  public Initializer() {
    logger.logInfo("initializing here called.");
    TestUtils.getInstance().initializeData();
  }  

  // this is the entrance point for my code
  @PostConstruct
  public void postInit() {
    for (int i = 0; i < 3; i++) {
        ProcessA process =
            new ProcessA("processA", properties);
        processAList.add(process);
        executorServiceProcessA.submit(process);         
    }       
  }

  @PreDestroy
  public void shutdown() {
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        for (ProcessA process : processAList) {
          process.shutdown();
        }
        executorServiceProcessA.shutdown();
        try {
          executorServiceProcessA.awaitTermination(1000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ex) {
          Thread.currentThread().interrupt();
        }
      }
    });
  }
}

我将只有一个Initializer 类，它将从其postInit 方法执行我的两个进程，然后从其shutdown 方法关闭这两个进程。

问题陈述：-

现在我该如何扩展我的设计，以便它可以有效地使用两个processes？每个进程都应该有自己的线程池，自己的属性对象，我应该能够从postInit 方法执行它们，然后稍后从shutdown 方法关闭。

最好和最有效的方法是什么？

Answer 1

您的代码明显违反了 DRY（不要重复自己）原则。

换句话说，在你的Process 和Main 类中有很多样板代码，可以通过使用抽象类来消除（或者如果你使用带有@987654323 的Java8，则使用接口） @ 方法）。

所以我创建了两个Process 和ProcessHandler 抽象类来重用每个进程和进程处理中通用的代码。 所以现在，您可以定义扩展 Process 类的 ProcessA、ProcessB 类和扩展 ProcessHandler 类的 ProcessHandlerA、ProcesshandlerB。

关键是这个方案可以扩展到任意数量的Process**，即遵循**OOP的Open/Closed原则。

您可以使用 cmets 参考以下代码：

流程类（摘要）：

public abstract Process implements Runnable {

        private final Properties props;
        private final String processName;

        public Process(String processName, Properties props) {
            this.processName = processName;
            this.props = props;
        }

        //this can also be a non abstract (reusable) method 
        // to eliminate boiler plate code (if any)
        public abstract void shutdown();
    }

ProcessA 类：

public class ProcessA extends Process {

          public ProcessA(String processName, Properties props) {
              super(processName, props);
          }

          @Override
          public void run() {
              //add run code here
          }

          @Override
          public void shutdown() {
              //shut down code
          }
    }

进程 B 类：

//与ProcessA类似，有B的具体细节

ProcessHandler 类（抽象）：

public abstract class ProcessHandler {

         private final ExecutorService executorServiceProcess;
         private final List<Process> processList;
         private int poolSize;

         protected ProcessHandler(int poolSize) {
             executorServiceProcess = Executors.newFixedThreadPool(poolSize);
             processList = new ArrayList<>();
             this.poolSize = poolSize;
         }

          public void postInit(Process process) {
             for (int i = 0; i < poolSize; i++) {
                    processList.add(process);
                    executorServiceProcess.submit(process);    
              }  
          }

         public void shutdown() {
                Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    for (Process process : processList) {
                      process.shutdown();
                    }
                    executorServiceProcess.shutdown();
                    try {
                      executorServiceProcess.
                        awaitTermination(1000, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException ex) {
                      Thread.currentThread().interrupt();
                    }
                  }
                });
         }
 }

ProcessHandlerA类：

public class ProcessHandlerA extends ProcessHandler {

          public ProcessHandlerA() {
              super(3);//configure pool size properly w.r.to ProcessA requirements
          }

          public void postInit() {
              ProcessA processA = new ProcessA("processA", properties);
              super(processA);
          }

          public void shutdown() {
              super.shutdown();
          }
     }

ProcessHandlerB 类：

//与ProcessHandlerA类似，B的具体细节