0

我正在尝试将线程池中的任务排队，以便在工作人员空闲时立即执行，我发现了各种示例，但在所有情况下，这些示例都已设置为为每个作业使用一个新的 Worker 实例，我想要持久的工人。

我正在尝试制作一个 ftp 备份工具，我可以让它工作，但由于单个连接的限制，它很慢。我理想中想要做的是有一个单一的连接来扫描目录并建立一个文件列表，然后四个工作人员下载所述文件。

这是我的 FTP 工作者的示例：

public class Worker implements Runnable {
  protected FTPClient _ftp;

  // Connection details
  protected String _host = "";
  protected String _user = "";
  protected String _pass = "";

  // worker status
  protected boolean _working = false;

  public Worker(String host, String user, String pass) {
    this._host = host;
    this._user = user;
    this._pass = pass;
  }

   // Check if the worker is in use
  public boolean inUse() {
    return this._working;
  }

  @Override
  public void run() {
    this._ftp = new FTPClient();
    this._connect();
  }

  // Download a file from the ftp server
  public boolean download(String base, String path, String file) {
    this._working   = true;
    boolean outcome = true;

    //create directory if not exists
    File pathDir = new File(base + path);
    if (!pathDir.exists()) {
      pathDir.mkdirs();
    }

    //download file
    try {
      OutputStream output = new FileOutputStream(base + path + file);
      this._ftp.retrieveFile(file, output);
      output.close();
    } catch (Exception e) {
      outcome = false;
    } finally {
      this._working = false;
      return outcome;
    }
  }

  // Connect to the server
  protected boolean _connect() {
    try {
      this._ftp.connect(this._host);
      this._ftp.login(this._user, this._pass);
    } catch (Exception e) {
      return false;
    }
    return this._ftp.isConnected();
  }

  // Disconnect from the server
  protected void _disconnect() {
    try {
      this._ftp.disconnect();
    } catch (Exception e) { /* do nothing */ }
  }
}

我希望能够Worker.download(...)在工作人员可用时调用队列中的每个任务，而不必为每次下载创建与 ftp 服务器的新连接。

任何帮助将不胜感激，因为我以前从未使用过线程，而且我现在正在兜圈子。

4

2 回答 2

4

这些示例已设置为为每个作业使用一个新的 Worker 实例，我想要持久的工作者。

这是一个常见问题，有几种不同的解决方案。您想要的是每个线程的一些上下文，而不是每个线程Runnable或Callable将提交给ExecutorService.

一种选择是拥有一个ThreadLocal可以创建您的ftp实例的选项。这不是最优的，因为当线程终止时没有简单的方法来关闭 ftp 连接。然后，您将通过限制线程池中运行的线程数来限制连接数。

我认为更好的解决方案是使用ExecutorServiceonly 来分叉您的工作线程。对于每个工人，向他们注入一个BlockingQueue他们都用来出列和执行他们需要做的任务的任务。这与内部使用的队列是分开的ExecutorService。然后，您会将任务添加到您的队列中，而不是添加到队列中 ExecutorService 。

private static final BlockingQueue<FtpTask> taskQueue
        = new ArrayBlockingQueue<FtpTask>();

因此，您的任务对象将具有以下内容：

public static class FtpTask {
     String base;
     String path;
     String file;
}

然后run()你的Worker类中的方法会做类似的事情：

public void run() {
    // make our permanent ftp instance
    this._ftp = new FTPClient();
    // connect it for the life of this thread
    this._connect();
    try {
        // loop getting tasks until we are interrupted
        // could also use volatile boolean !shutdown
        while (!Thread.currentThread().isInterrupted()) {
            FtpTask task = taskQueue.take();
            // if you are using a poison pill
            if (task == SHUTDOWN_TASK) {
                break;
            }
            // do the download here
            download(task.base, task.path, task.file);
        }
    } finally {
        this._disconnect();
    }
}

同样，您通过限制线程池中运行的线程数来限制连接数。

我理想中想要做的是有一个单一的连接来扫描目录并建立一个文件列表，然后四个工作人员下载所述文件。

我将有一个Executors.newFixedThreadPool(5);并添加一个执行扫描/构建的线程和 4 个正在执行下载的工作线程。BlockingQueue当工作线程从同一个队列中取出时，扫描线程将被放入。

于 2013-10-17T15:33:55.153 回答
2

我建议根据要求选择具有核心大小和 maxpoolsize 的 ThreadPoolexecutor。在这种情况下，还要使用链接阻塞队列，它将以 FIFO 方式在其中执行您的任务。

一旦 Thread(worker) 空闲，就会从队列中挑选任务并执行。

查看 ThreadPoolExecutor 的详细信息。如果您在执行 ThreadPoolexecutor 时遇到任何问题，请告诉我。

于 2013-10-17T15:29:45.663 回答