Java 线程竞争条件与 BlockingQueue

Question

我的 Java 代码中使用 BlockingQueue 时存在潜在的竞争条件，我想知道如何修改代码来避免这种情况:

private static BlockingQueue<FileToFTP> ftpQueue = new LinkedBlockingQueue<FileToFTP>();

private static boolean shuttingDown = false;

private static FileToFTP pendingFile = null;
private static int uploadsPending = 0;

private static Thread ftpThread = new Thread(new Runnable() { 
            public void run() {
                try {
                    for(;;) {
                        FileToFTP f2f = ftpQueue.take();    // this blocks until there is something to take
                        // FIXME: If the main thread takes over right 
                        // at this instant, then uploadsRemaining() 
                        // is not correct!!
                        uploadsPending = 1;
                        int qs = ftpQueue.size();
                        logIt(qs, f2f);
                        pendingFile = f2f;
                        if(!doUploadFile(f2f.getPath(), f2f.getFilename(), f2f.getRenameTo(), f2f.isBinary())) {
                            if(shuttingDown) {
                                log.info("Upload " + f2f + " failed!");
                            } else {
                                ftpQueue.offer(f2f);    // put it back on to retry later
                            }
                            uploadsPending = 0;
                        } else {
                            pendingFile = null;
                            uploadsPending = 0;
                        }
                        if(qs == 0) logIt("");
                    }
                } catch (InterruptedException consumed) {
                    // Allow thread to exit
                }
            }
        });

public static int uploadsRemaining() {
    return ftpQueue.size() + uploadsPending;
}

请参阅代码中的“FIXME”注释。谢谢!!

Answer 1

也许我误解了你想要的，但听起来你最好使用 ExecutorService 实际运行事物。您可以使用 Exectors.newSingleThreadExecutor() 创建它们或 Executors.newFixedThreadPool(2) (2 是它使用的线程数的示例)。

然后您可以 .execute 或 .submit Runnable上 ExecutorService 。 submit将返回 Future<T> 对象可用于跟踪提交给 ExecutorService 的特定作业的状态。

话虽如此，您可能需要创建一个新类来执行此操作，如 Runnable/Callable类需要有 FileToFTP其中的变量。 (旁注:我尽量避免内部类，所以如果您想知道为什么我不建议这样做......)

问题就变成了“我如何知道有多少文件正在等待处理？”不幸的是，唯一简单的方法是使其成为另一个类的静态属性......使用静态 get/set 方法或作为公共(public)/ protected 属性。安 AtomicInteger 是理想的选择，因此您可能需要考虑在调用类中使用它作为 protected 静态类。它具有专用的递增/递减命令以及调整值和返回值的命令。