Java中如何进行线程限制

Question

假设我有 1000 个文件要读取，并且由于某些限制，我想并行读取最多 5 个文件。而且，一旦其中一个完成，我就想要一个新的开始。

我有一个主函数，它有文件列表，并且每当一个线程完成时我都会尝试更改计数器。但它不起作用！

有什么建议吗？

以下是main函数循环

for (final File filename : folder.listFiles()) {

    Object lock1 = new Object();
    new myThread(filename, lock1).start();
    counter++;
    while (counter > 5);
}

Answer 1

像这样生成线程并不是正确的方法。使用 ExecutorService 并将池指定为 5。将所有文件放入 BlockingQueue 或其他线程安全集合之类的东西中，所有正在执行的文件都可以poll() 随意。

public class ThreadReader {

    public static void main(String[] args) {
        File f = null;//folder
        final BlockingQueue<File> queue = new ArrayBlockingQueue<File>(1000);
        for(File kid : f.listFiles()){
            queue.add(kid);
        }

        ExecutorService pool = Executors.newFixedThreadPool(5);

        for(int i = 1; i <= 5; i++){
            Runnable r = new Runnable(){
                public void run() {
                    File workFile = null;
                    while((workFile = queue.poll()) != null){
                        //work on the file.
                    }
                }
            };
            pool.execute(r);
        }
    }
}

Answer 2

您可以使用 ExecutorService 作为线程池和队列。

ExecutorService pool = Executors.newFixedThreadPool(5);
File f = new File(args[0]);

for (final File kid : f.listFiles()) {
    pool.execute(new Runnable() {
        @Override
        public void run() {
            process(kid);
        }
    });
}
pool.shutdown();
// wait for them to finish for up to one minute.
pool.awaitTermination(1, TimeUnit.MINUTES);

Answer 3

凯拉的回答中的方法是正确的。使用 Java 类库提供的执行器类，而不是从头开始自己实现线程池（很糟糕）。

---

但我认为讨论您问题中的代码以及为什么它不起作用可能会有用。 （我已经尽力填写了您遗漏的一些部分...）

public class MyThread extends Thread {

    private static int counter;

    public MyThread(String fileName, Object lock) {
        // Save parameters in instance variables
    }

    public void run() {
        // Do stuff with instance variables
        counter--;
    }

    public static void main(String[] args) {
        // ...
        for (final File filename : folder.listFiles()) {
            Object lock1 = new Object();
            new MyThread(filename, lock1).start();
            counter++;
            while (counter > 5);
        }
        // ...
    }
}

好吧，那么这有什么问题呢？为什么不起作用？

第一个问题是，在 main 中，您正在读取和写入 counter 而不进行任何同步。我假设它也由工作线程更新 - 否则代码没有任何意义。因此，这意味着主线程很可能看不到子线程所做的更新的结果。换句话说，while (counter > 5);可能是一个无限循环。 （事实上​​，这很有可能。JIT 编译器可以生成这样的代码，其中 counter > 5 只是测试在前面的 counter++; 语句。

第二个问题是您的 while (counter > 5); 循环非常浪费资源。 ..并且它可能会执行数十亿次第二...全力运行一个处理器（核心） 如果您打算使用低级原语实现此类内容，则应该使用 Java 的 Object.wait() 和 Object.notify() 方法；例如主线程等待，每个工作线程通知。

Answer 4

无论您使用什么方法来创建新线程，递增全局计数器，在线程创建周围添加条件语句，如果达到限制，则不要创建新线程，也许将文件推送到队列（一个列表） ？），然后您可以在创建线程后添加另一个条件语句，如果队列中有项目，则首先处理这些项目。

Answer 5

我们也可以使用信号量类。
因此 Semaphore 类将限制代码块上一次执行的线程数量。它有像 aquire()、release() 这样的方法来获取和释放锁这种机制