了解 Java ExecutorService

Question

我正在尝试学习如何使用 Java 的 executorservice，

我正在阅读以下讨论Java thread simple queue

这里有一个示例

ExecutorService service = Executors.newFixedThreadPool(10);
// now submit our jobs
service.submit(new Runnable() {
    public void run() {
    do_some_work();
   }
});
// you can submit any number of jobs and the 10 threads will work on them
// in order
...
// when no more to submit, call shutdown
service.shutdown();
// now wait for the jobs to finish
service.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

我尝试实施此解决方案，为此我创建了一个表单并放置了启动和停止按钮，但我面临的问题是，如果我在启动按钮上调用此过程，它将挂起完整的表单，我们需要等待直到所有过程完成。

我也尝试阅读以下https://www3.ntu.edu.sg/home/ehchua/programming/java/J5e_multithreading.html

但是直到现在我还无法理解如何使它工作，因为在单击开始按钮后，我应该可以恢复访问权限，假设我想停止该过程。

有人可以指导我正确的方向吗？

谢谢

为了让我的情况更清楚，我正在添加我正在测试的代码。

问题

1) 完整的表格在程序执行时保持冻结状态。 2) 进度条不起作用，只有在所有进程完成后才会显示状态。

private void btnStartActionPerformed(java.awt.event.ActionEvent evt) {                                         
  TestConneciton();

}                                        

private void btnStopActionPerformed(java.awt.event.ActionEvent evt) {                                        
    flgStop = true;
}   

   private static final int MYTHREADS = 30;
private boolean flgStop = false;
public  void TestConneciton() {
    ExecutorService executor = Executors.newFixedThreadPool(MYTHREADS);
    String[] hostList = { "http://crunchify.com", "http://yahoo.com",
            "http://www.ebay.com", "http://google.com",
            "http://www.example.co", "https://paypal.com",
            "http://bing.com/", "http://techcrunch.com/",
            "http://mashable.com/", "http://thenextweb.com/",
            "http://wordpress.com/", "http://wordpress.org/",
            "http://example.com/", "http://sjsu.edu/",
            "http://ebay.co.uk/", "http://google.co.uk/",
            "http://www.wikipedia.org/",
            "http://en.wikipedia.org/wiki/Main_Page" };

    pbarStatus.setMaximum(hostList.length-1);
    pbarStatus.setValue(0);
    for (int i = 0; i < hostList.length; i++) {

        String url = hostList[i];
        Runnable worker = new MyRunnable(url);
        executor.execute(worker);
    }
    executor.shutdown();
    // Wait until all threads are finish
//        while (!executor.isTerminated()) {
// 
//        }
    System.out.println("\nFinished all threads");
}

public  class MyRunnable implements Runnable {
    private final String url;

    MyRunnable(String url) {
        this.url = url;
    }

    @Override
    public void run() {

        String result = "";
        int code = 200;
        try {
            if(flgStop == true)
            {
                //Stop thread execution
            }
            URL siteURL = new URL(url);
            HttpURLConnection connection = (HttpURLConnection) siteURL
                    .openConnection();
            connection.setRequestMethod("GET");
            connection.connect();

            code = connection.getResponseCode();
            pbarStatus.setValue(pbarStatus.getValue()+1);
            if (code == 200) {
                result = "Green\t";
            }
        } catch (Exception e) {
            result = "->Red<-\t";
        }
        System.out.println(url + "\t\tStatus:" + result);
    }
}

Answer 1

根据ExecutorService API，此阻止：

service.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

API 报价：

在关闭请求后阻塞，直到所有任务完成执行，或发生超时，或当前线程被中断，以先发生者为准。

如果您不希望它阻塞当前线程，那么也许您应该在不同的线程中调用它。另外，您的是否是 Swing 应用程序，然后考虑使用 SwingWorker，我相信它会在“幕后”使用 ExecutorServices。

---

根据您最新的代码，我会使用

• 一个用于管理所有后台线程的 SwingWorker。
• 我会给 SwingWorker 一个 ExecutorService
• 还有一个使用 ExecutorService 初始化的 ExecutorCompletionService，因为这可以让我在任务结果完成时获取它们
• 我会用 Callables 填充它，不是 Runnables，因为这将允许任务返回一些东西，也许是一个表示进度的字符串。
• 我将 SwingWorker 的进度属性设置为 (100 * taskCount) / totalTask​​Count 并让我的 JProgressBar 从 0 变为 100。
• 然后我将使用 SwingWorker 的发布/处理方法对提取可调用对象返回的字符串。
• 我会使用 PropertyChangeListener 在我的 GUI 中收听 SwingWorker 的进度
• 然后在此侦听器中更改 GUI。
• 我会将if (code == 200) { 更改为if (code == HttpURLConnection.HTTP_OK) { 以避免出现幻数。
• JButton 的 Action 会自行禁用，然后创建一个新的 SwingWorker 对象，将 worker 的 ProperChangeListener 添加到 worker，然后执行 worker。

---

例如

import java.awt.Font;
import java.awt.event.ActionEvent;
import java.awt.event.KeyEvent;
import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.swing.AbstractAction;
import javax.swing.Action;
import javax.swing.DefaultListModel;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JList;
import javax.swing.JPanel;
import javax.swing.JProgressBar;
import javax.swing.JScrollPane;
import javax.swing.SwingUtilities;
import javax.swing.SwingWorker;

@SuppressWarnings("serial")
public class SwingExecutorCompletionService extends JPanel {
   public static final int MYTHREADS = 10;
   private static final int LIST_PROTOTYPE_SIZE = 120;
   private static final String LIST_PROTOTYPE_STRING = "%" + LIST_PROTOTYPE_SIZE + "s";
   public static final String[] HOST_LIST = { 
      "http://crunchify.com",
      "http://yahoo.com", 
      "http://www.ebay.com", 
      "http://google.com",
      "http://www.example.co", 
      "https://paypal.com", 
      "http://bing.com/",
      "http://techcrunch.com/", 
      "http://mashable.com/",
      "http://thenextweb.com/", 
      "http://wordpress.com/",
      "http://wordpress.org/", 
      "http://example.com/", 
      "http://sjsu.edu/",
      "http://ebay.co.uk/", 
      "http://google.co.uk/",
      "http://www.wikipedia.org/", 
      "http://en.wikipedia.org/wiki/Main_Page" };

   private JProgressBar pbarStatus = new JProgressBar(0, 100);
   private JButton doItButton = new JButton(new DoItAction("Do It", KeyEvent.VK_D));
   private DefaultListModel<String> listModel = new DefaultListModel<>();
   private JList<String> resultList = new JList<>(listModel);

   public SwingExecutorCompletionService() {
      resultList.setVisibleRowCount(10);
      resultList.setPrototypeCellValue(String.format(LIST_PROTOTYPE_STRING, ""));
      resultList.setFont(new Font(Font.MONOSPACED, Font.PLAIN, 12));

      add(pbarStatus);
      add(doItButton);
      add(new JScrollPane(resultList, JScrollPane.VERTICAL_SCROLLBAR_ALWAYS, 
            JScrollPane.HORIZONTAL_SCROLLBAR_AS_NEEDED));
   }

   public void addToCompletionList(String element) {
      listModel.addElement(element);
   }

   public void setStatusValue(int progress) {
      pbarStatus.setValue(progress);
   }

   class DoItAction extends AbstractAction {
      public DoItAction(String name, int mnemonic) {
         super(name);
         putValue(MNEMONIC_KEY, mnemonic);
      }

      @Override
      public void actionPerformed(ActionEvent e) {
         setEnabled(false);
         DoItWorker worker = new DoItWorker(HOST_LIST, MYTHREADS);
         SwingExecutorCompletionService gui = SwingExecutorCompletionService.this;
         PropertyChangeListener workerListener = new WorkerChangeListener(gui, this);
         worker.addPropertyChangeListener(workerListener);
         worker.execute();
      }
   }

   private static void createAndShowGui() {
      SwingExecutorCompletionService mainPanel = new SwingExecutorCompletionService();

      JFrame frame = new JFrame("Swing ExecutorCompletionService");
      frame.setDefaultCloseOperation(JFrame.DISPOSE_ON_CLOSE);
      frame.getContentPane().add(mainPanel);
      frame.pack();
      frame.setLocationByPlatform(true);
      frame.setVisible(true);
   }

   public static void main(String[] args) {
      SwingUtilities.invokeLater(new Runnable() {
         public void run() {
            createAndShowGui();
         }
      });
   }
}

class MyCallable implements Callable<String> {
   private static final String RED = "->Red<-";
   private static final String GREEN = "Green";
   private final String url;
   private volatile boolean flgStop;

   MyCallable(String url) {
      this.url = url;
   }

   @Override
   public String call() throws Exception {
      String result = "";
      int code = HttpURLConnection.HTTP_OK;
      try {
         // if(flgStop == true)
         if (flgStop) {
            // Stop thread execution
         }
         URL siteURL = new URL(url);
         HttpURLConnection connection = (HttpURLConnection) siteURL
               .openConnection();
         connection.setRequestMethod("GET");
         connection.connect();

         code = connection.getResponseCode();

         // No don't set the prog bar in a background thread!
         // !! pbarStatus.setValue(pbarStatus.getValue()+1); 
         // avoid magic numbers
         if (code == HttpURLConnection.HTTP_OK) {
            result = GREEN;
         }
      } catch (Exception e) {
         result = RED;
      }
      return String.format("%-40s %s", url + ":", result);
   }

}

class WorkerChangeListener implements PropertyChangeListener {
   private Action action;
   private SwingExecutorCompletionService gui;

   public WorkerChangeListener(SwingExecutorCompletionService gui, Action button) {
      this.gui = gui;
      this.action = button;
   }

   @Override
   public void propertyChange(PropertyChangeEvent evt) {
      DoItWorker worker = (DoItWorker)evt.getSource();
      if (evt.getNewValue() == SwingWorker.StateValue.DONE) {
         action.setEnabled(true);
         try {
            worker.get();
         } catch (InterruptedException e) {
            e.printStackTrace();
         } catch (ExecutionException e) {
            e.printStackTrace();
         }
      } else if (DoItWorker.INTERMEDIATE_RESULT.equals(evt.getPropertyName())) {
         gui.addToCompletionList(evt.getNewValue().toString());
      } else if ("progress".equals(evt.getPropertyName())) {
         gui.setStatusValue(worker.getProgress());
      }
   }
}

class DoItWorker extends SwingWorker<Void, String> {
   public static final String INTERMEDIATE_RESULT = "intermediate result";
   private static final long TIME_OUT = 5;
   private static final TimeUnit UNIT = TimeUnit.MINUTES;

   private String intermediateResult;
   private ExecutorService executor;
   private CompletionService<String> completionService;
   private String[] hostList;

   public DoItWorker(String[] hostList, int myThreads) {
      this.hostList = hostList;
      executor = Executors.newFixedThreadPool(myThreads);
      completionService = new ExecutorCompletionService<>(executor);
   }

   @Override
   protected Void doInBackground() throws Exception {
      for (int i = 0; i < hostList.length; i++) {

         String url = hostList[i];
         Callable<String> callable = new MyCallable(url);
         completionService.submit(callable);
      }
      executor.shutdown();
      for (int i = 0; i < hostList.length; i++) {
         String result = completionService.take().get();
         publish(result);
         int progress = (100 * i) / hostList.length;
         setProgress(progress);
      }
      executor.awaitTermination(TIME_OUT, UNIT);
      setProgress(100);
      return null;
   }

   @Override
   protected void process(List<String> chunks) {
      for (String chunk : chunks) {
         setIntermediateResult(chunk);
      }
   }

   private void setIntermediateResult(String intermediateResult) {
      String oldValue = this.intermediateResult;
      String newValue = intermediateResult;
      this.intermediateResult = intermediateResult;
      firePropertyChange(INTERMEDIATE_RESULT, oldValue, newValue);
   }

}

看起来和运行起来像：

Answer 2

如果您使用JButton 和ExecutorService，那么您可能应该创建一个新线程并释放事件调度线程 (EDT)：

button.setActionListener(new Action() {
  public void actionPerformed(ActionEvent e) {
    button.setEnabled(false); // disable the button 
    new Thread(new Runnable() {
      public void run() {
        ... your code ...
        button.setEnabled(true);
      }
    }).start();
  } 
});

就像 Hovercraft Full of Eels 说的，awaitTermination 是一个阻塞操作。

在 Swing 的情况下，您可能在 Action 中执行此操作（如我的示例），并且您正在阻止 EDT 执行各种操作，例如响应用户输入:)

注意：ExecutorService 有一个不错的invokeAll，这可能比使用awaitTermination 更有用。这也会阻塞，如果需要，您仍然需要在另一个线程中执行您的操作。

Answer 3

如果您想取消已经开始的作业，那么您必须使用Callable 而不是Runnable。当你提交一份工作时，你会得到一个Future，你可以调用cancel()。

public static void main(String[] args) {
  ExecutorService es = Executors.newFixedThreadPool(10);
  Callable<Integer> callable1 = new CallableImpl();
  Future<Integer> future1 = es.submit(callable1);

  // if you decide to cancel your task gracefully
  future1.cancel()
  ...

然后由您在 Callable 的实现中处理 ThreadInterrupt。

public class CallableImpl implements Callable<Integer> {
@Override
public Integer call() throws Exception {
     try {
        while(true) {               
            // do something

            if(Thread.currentThread().isInterrupted()) {
                System.out.println("detected interrupt flag");
                break;
            }
        }
    }
    catch(InterruptedException ie) {
        System.out.println("interrupted");
    }

@Hovercraft 可能是对的，如果您正在编写 Swing 应用程序，那么您想要使用 SwingWorker。