java - 为什么并发线程限制没有按预期工作？

Question

虽然有类似的问题，但我找不到与我得到的类似的例子。我非常感谢任何帮助理解我的实现错误的地方。

我正在尝试做的事情:

我有一个主类 Driver ，它可以实例化未知数量的线程。每个线程调用一个 singleton应该模拟“假”文件传输操作的类。我遇到的问题是我需要将并发传输限制为 2 次传输，无论并发请求的数量如何。

我尝试解决问题的方法是添加每个新的 Thread在ConcurrentLinkedQueue并使用 Executors.newFixedThreadPool(POOL_SIZE) 对其进行管理将并发线程限制为 2。对于每次交互 - 我使用 pool.submit 从池中轮询新线程.

我遇到的问题是我的输出是这样的:[线程 1]、[线程 1、线程 2]、[线程 1、线程 2、线程 3]...

虽然它应该是:[线程1，线程2]，[线程3，线程4]

为什么限制在这里不起作用？

我的实现:

复印机 - 这是我的singleton类。

public class Copier {

    private final int POOL_SIZE = 2;
    private static volatile Copier instance = null;
    private Queue<Reportable> threadQuere = new ConcurrentLinkedQueue();
    private static FileCopier fileCopier = new FileCopier();

    private Copier() {

    }

    public static Copier getInstance() {
        if (instance == null) {
            synchronized (Copier.class) {
                if (instance == null) {
                    instance = new Copier();
                }
            }
        }
        return instance;
    }

    public void fileTransfer(Reportable reportable) {
        threadQuere.add(reportable);

        ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
        for (int i=0; i < threadQuere.size(); i++) {
            System.out.println("This is the " + (i+1) + " thread");
            pool.submit(new CopyThread());
        }
        pool.shutdown();
        try {
            pool.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

CopyThread - 表示一个线程类

    public class CopyThread implements Reportable, Runnable {

    private static FileCopier fileCopier = new FileCopier();

    @Override
    public void report(String bitrate) {
        System.out.println(bitrate);
    }

    @Override
    public void run() {
        synchronized(fileCopier) {
            long startTime = System.nanoTime();
            long bytes = fileCopier.copyFile();
            long endTime = System.nanoTime();

            double duration = (double)(endTime - startTime) / 1000000000; // get in seconds
            double bytesInMegas = (double) bytes / 1000000;

            report(bytesInMegas + "MB were transferred in " + duration + " seconds");
        }
    }
}

司机 - 我的main我在哪里创建所有线程的类

    public class Driver {
    public static void main(String[] args) {
        Copier copier = Copier.getInstance();
        CopyThread copyThread1 = new CopyThread();
        CopyThread copyThread2 = new CopyThread();
        CopyThread copyThread3 = new CopyThread();
        CopyThread copyThread4 = new CopyThread();

        copier.fileTransfer(copyThread1);
        copier.fileTransfer(copyThread2);
        copier.fileTransfer(copyThread3);
        copier.fileTransfer(copyThread4);
        int q = 0;

    }
}

Answer 1

一个更简单的解决方案是 Semaphore有 2 个许可证。

这确保“外部”线程也无法绕过限制，因为您的解决方案预计同时执行的任务会受到线程池大小的限制。

当一个工具就足够时，您的解决方案会使用多个并发工具。您的 DCL 单例也有点过时了。