java - 如何确保每个线程工作在不同的进程对象上以避免线程安全问题？

Question

我有两个流程，如下所示。我的每个进程都有 run 和 shutdown 方法

Process processA = new ProcessA("processA", getProcessAProperties());
Process processB = new ProcessB("processB", getProcessBProperties());

• 我希望每个进程有不同的线程池配置，以便 ProcessA 在自己的线程池中运行，而 ProcessB 在自己的线程池中运行，彼此独立。
• 而且我无法在其自己的线程池的每个线程之间共享 Process 对象。

下面是我如何使用自己的线程池运行 ProcessA 的简单示例。有三个线程，每个线程都有自己的 Process 对象来处理。现在我想以更通用的方式扩展它，以便它可以适用于我的两个流程，如上所示。

public static void main(String[] args) {
  int numberOfThreads = 3;
  ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);

  final List<Process> processes = new ArrayList<>();
  for (int i = 0; i < numberOfThreads; i++) {
    // each thread works on different Process object
    Process processA = new ProcessA("processA", getProcessAProperties());
    processes.add(processA);
    executor.submit(processA);
  }

  Runtime.getRuntime().addShutdownHook(new Thread() {
    @Override
    public void run() {
      for (Process process : processes) {
        process.shutdown();
      } 
      executor.shutdown();
      try {
        executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        e.printStackTrace;
      }
    }
  });
}

因此我创建了一个进程处理程序来以通用方式解决上述问题，但这里存在线程安全问题:

public final class ProcessHandler {
  private final ExecutorService executorServiceProcess;
  private final Process process;
  private final Thread shutdownHook = new Thread() {
    @Override
    public void run() {
      process.shutdown();
      executorServiceProcess.shutdown();
    }
  };

  // in this constructor my code is reusing the same 
  // process instance for each thread in the pool 
  // which is a problem for my application, how to fix this?
  public ProcessHandler(Process process, int poolSize) {
    this.executorServiceProcess = Executors.newFixedThreadPool(poolSize);
    this.process = process;
    Runtime.getRuntime().addShutdownHook(shutdownHook);     
    for (int i = 0; i < poolSize; i++) {
      executorServiceProcess.submit(process);
    }
  }

  public void shutdown() {
    Runtime.getRuntime().removeShutdownHook(shutdownHook);
    shutdownHook.start();
    try {
      shutdownHook.join();
    } catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
  }
}

这就是我的主要方法现在的样子:

public static void main(String[] args) {

    Process processA = new ProcessA("processA", getProcessAProperties());
    Process processB = new ProcessB("processB", getProcessBProperties());

    // processA will run with three threads in its own thread pool
    ProcessHandler processHandlerA = new ProcessHandler (processA, 3);
    // processB will run with two threads in its own thread pool
    ProcessHandler processHandlerB = new ProcessHandler (processB, 2);

    // now I can call shutdown on them
    processHandlerA.shutdown();
    processHandlerB.shutdown();
}

正如您在上面的 ProcessHandler 构造函数中看到的，我为池中的每个线程重用相同的流程实例，这不是我想要做的。我希望每个线程都在 Process 对象的不同实例上工作，就像在 ProcessA 的第一个 main 方法中一样，每个线程都在不同的 Process 对象上工作。

解决这个设计问题的最佳方法是什么？我也愿意重新设计我的 ProcessHandler，以正确的方式解决这个问题。

Answer 1

也许尝试这样的事情:

// Replace Process process by a list of Process
List<Process> processes = new ArrayList<Process>();

private final Thread shutdownHook = new Thread() {
    @Override
    public void run() {
      for (Process process : processes)
        process.shutdown();
      executorServiceProcess.shutdown();
    }
};

public ProcessHandler(Process process, int poolSize) {
    this.executorServiceProcess = Executors.newFixedThreadPool(poolSize);
    Runtime.getRuntime().addShutdownHook(shutdownHook);     
    for (int i = 0; i < poolSize; i++) {
      // Get a deep copy of the process
      Process p = process.clone();
      processes.add(p);
      executorServiceProcess.submit(p);
    }
}

另一种不使用克隆方法的方法是抽象 Process 并基于它们添加两个构造函数，不要忘记根据这些元素调整您的代码

public abstract class AProcess extends Process {

    private String name;
    private Properties properties;
    public AProcess(String name, Properties properties)
    {
        this.setName(name);
        this.setProperties(properties);
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Properties getProperties() {
        return properties;
    }

    public void setProperties(Properties properties) {
        this.properties = properties;
    }
}

您现在必须实现 ProcessA 和 ProcessB :

public class ProcessA extends AProcess {

    public ProcessA(String name, Properties properties)
    {
        super(name, properties);
    }
}

现在创建您的 ProcessHandler :

public final class ProcessHandler {
  private final ExecutorService executorServiceProcess;
  private final List<AProcess> processes = new ArrayList<AProcess>();
  private final Thread shutdownHook = new Thread() {
    @Override
    public void run() {
      for (AProcess process : processes)
          process.shutdown();
      executorServiceProcess.shutdown();
    }
  };

  public ProcessHandler(ProcessA process, int poolSize) {
    this.executorServiceProcess = Executors.newFixedThreadPool(poolSize);
    Runtime.getRuntime().addShutdownHook(shutdownHook);     
    for (int i = 0; i < poolSize; i++) {
      ProcessA p = new ProcessA(process.getName(), process.getProperties());
      processes.add(p);
      executorServiceProcess.submit(p);
    }
  }

  public ProcessHandler(ProcessB process, int poolSize) {
    this.executorServiceProcess = Executors.newFixedThreadPool(poolSize);
    Runtime.getRuntime().addShutdownHook(shutdownHook);     
    for (int i = 0; i < poolSize; i++) {
      ProcessB p = new ProcessB(process.getName(), process.getProperties());
      processes.add(p);
      executorServiceProcess.submit(p);
    }
  }

  public void shutdown() {
    Runtime.getRuntime().removeShutdownHook(shutdownHook);
    shutdownHook.start();
    try {
      shutdownHook.join();
    } catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
  }
}

并像这样使用ProcessHandler processHandlerA = new ProcessHandler(processA, 3);