如何在不阻塞的情况下避免并发执行一个耗时的任务？答案

【问题标题】：How to avoid concurrent execution of a time-consuming task without blocking?如何在不阻塞的情况下避免并发执行一个耗时的任务？
【发布时间】：2014-06-03 03:02:36
【问题描述】：

我想有效地避免在大量多线程环境中并发执行耗时的任务，而不会在另一个线程已经在运行任务时让线程等待锁定。相反，在那种情况下，我希望它们能够优雅地失败（即跳过它执行任务的尝试）尽可能快。换句话说：当任务已经在进行中时，我需要尝试再次启动任务以立即退出，最好不要同步成本。

为了说明这个不安全（有竞争条件！）代码的想法：

private static boolean running = false;

public void launchExpensiveTask() {
    if (running) return; // Do nothing

    running = true;
    try {
        runExpensiveTask();
    } finally {
        running = false;
    }
}

我虽然关于使用双重检查锁定的变体（考虑到running 是一个原始的 32 位字段，因此是原子的，即使对于低于 5 的 Java 也可以正常工作，而不需要 volatile）。它可能看起来像这样：

private static boolean running = false;
private static Object execLock = new Object();

public void launchExpensiveTask() {
    if (running) return; // Do nothing

    synchronized (execLock) {
        if (running) return;

        running = true;
        try {
            runExpensiveTask();
        } finally {
            running = false;
        }
    }
}

也许我也应该使用该字段的本地副本（现在不确定，请告诉我）。

但后来我意识到无论如何我都会以一个内部同步块结束，它仍然可以在监视器入口处保持一个具有正确时间的线程，直到原始执行程序离开临界区（我知道几率通常很小，但在这个如果我们正在考虑在几个线程中竞争这个长期运行的资源）。

那么，你能想出更好的方法吗？

编辑：我之前省略了部分上下文，为了正确起见，我需要在执行期间保持一个锁，以保持其他方法试图改变一些内部共享状态。公平地说，到目前为止，我支持了有用的答案，包括两种情况：开始任务后需要和不需要锁。

【问题讨论】：

在 JVM 上进行并发编程的更好方法通常涉及Akka。离开陈旧的synchronized 垃圾，加入未来吧:)
是的，我真的很喜欢它，甚至更喜欢 Scala，但这个项目的决定权不在于我。无论如何，有价值的建议，谢谢。
我有一个数据流库，除其他外，它检查任务是否已经在运行。它使用 ReentrantLock 进行同步。在低端计算机上，同步延迟大约为 0.5-1 微秒。使用同步运算符只会稍微差一点。你真的需要更快的速度吗？ Akka 也有类似的表现。
嗨@Alexei。我想我知道你在想什么。为了帮助其他没有与我相同的性能限制的人，我在here 进行了澄清，并为他们提供了更好的解决方案。

标签： java concurrency synchronization

【解决方案1】：

我认为这更有意义：

 static volatile Boolean running = false;

    public static void launchTask()
    {
        synchronized(running)
        {
            if(running) return;
            running = true;
        }
            //DOSTUFF
            running = false;
    }

因为你真的只需要在设置布尔值时同步：如果多个线程同时询问，第一个将运行设置为true，其余的都将返回。

但是，您的设计可能会有更好的整体模式。如果线程向队列提交请求，（一个 ExecutorService？）得到 Future 或 ListenableFuture（来自 Guava）对象，然后继续做其他事情，直到 future 完成计算？

【讨论】：

您提出了一个很好的观点，即不要同步超过需要的部分（我将在问题中澄清这一点）。为你点赞:)
我担心对running 的最后一次写入不同步，特别是考虑到引擎盖下的自动装箱。你确定这段代码是线程安全的吗？
我不是 100% 肯定，但我不明白怎么可能。一次只有一个线程可以通过同步块，因此线程可以安全地将 running 设置为 false。即使当初始线程将 running 设置为 false 时另一个线程进入同步块也没关系：它最后将 running 设置为 false，任务完成，另一个线程可以重新启动它。如果担心自动装箱，您可以随时将 running 设置为 Boolean.FALSE
因为running = false; 与“DOSTUFF”一起在一个不同步的区域内，所以没有什么可以禁止优化编译器在“DOSTUFF”之前重新排序该句子（它们之间没有发生之前的关系）。然后另一个线程可以通过同步块，而第一个线程还没有完成“DOSTUFF”，因此不互斥。
有理由使用布尔值和单独的锁对象，而不是同时承担这两种职责的布尔值：经验表明自动装箱是邪恶的（1，2）。在这种特殊情况下，它给人的印象是，当您执行running = true 时，您正在修改字段的值，而实际上您正在用另一个实例替换它……再见了锁！

【解决方案2】：

忽略我的其他答案。但你要找的是这个。 http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/Semaphore.html

通过使用信号量。考虑信号量的最简单方法是将其视为允许获取 n 个单元并提供获取和释放机制的抽象。 TryAcquire 是关键，因为根据 java 的文档 - 只有在调用时可用的情况下，才能从此信号量获取许可。自己尝试一下。

    private Semaphore semaphore = new Semaphore(1);

    public void launchExpensiveTask() {
        if (semaphore.tryAcquire()) {
            try {
               runExpensiveTask();
            } finally {
               semaphore.release();
            }
        }

    }

【讨论】：

我喜欢您的代码的简单性，但不足以满足我的需求。此外，这里没有真正的理由更喜欢二进制信号量而不是 ReentrantLock（请参阅my answer），因为您在这里使用二进制信号量所做的所有事情都可以由 ReentrantLock 完成，而且更安全。
关于我之前关于二进制信号量的陈述，请看this answer。
如果可以问你还有什么需求？
抱歉耽搁了，我有点错过了你的最后评论。我需要任何新的尝试在已经运行的任务时尽快退出，最好没有同步成本。谢谢！
是的，使用我的示例不会在任务尝试获取时延迟任务。

【解决方案3】：

请记住，对于大多数人来说，以下解决方案会更可取：

private static Lock execLock = new ReentrantLock();

public void launchExpensiveTask() {
    if (!execLock.tryLock()) return; // skip if already running

    try {
        runExpensiveTask();
    } finally {
        lock.unlock();
    }
}

请注意，像这个问题中公开的这种特殊情况非常罕见，通常同步原语的性能绰绰有余，代码越简单越好。 避免过早的优化，除非你确定它不适合你，否则使用它。

【讨论】：

【解决方案4】：

借助Lock#tryLock()（从Java 5 开始提供API），我们可以做到非阻塞：

private static boolean running = false;
private static Lock execLock = new ReentrantLock();

public void launchExpensiveTask() {
    if (running) return; // fast exit without sync

    if (!execLock.tryLock()) return; // quit if lock is not free

    try {
        running = true;
        runExpensiveTask();
    } finally {
        running = false;
        execLock.unlock();
    }

}

如果您在任务执行过程中不需要持有锁，请查看以下代码：

private static boolean running = false;
private static Object execLock = new Object();

private boolean start() {
    synchronized (execLock) {
        boolean ret = running;
        running = true;
        return ret;
    }
}

private void end() {
    synchronized (execLock) {
        running = false;
    }
}

public void launchExpensiveTask() {
    if (running) return; // fast exit without sync

    if (start()) return; // already running, do nothing

    try {
        runExpensiveTask();
    } finally {
        end();
    }
}

【讨论】：

【解决方案5】：

更新：与问题所有者讨论后，这是最终提出的解决方案，附代码：

package toys;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class TwoQueues {

//tweak it for your purpose.
private final int CPU_COUNT = 4;
private BlockingQueue<Runnable> lightTaskQueue = new LinkedBlockingDeque<Runnable>();
private ThreadPoolExecutor lightExecutor = new ThreadPoolExecutor(CPU_COUNT, CPU_COUNT, 60L, TimeUnit.SECONDS, lightTaskQueue);
private BlockingQueue<Runnable> heavyTaskQueue = new LinkedBlockingDeque<Runnable>();
private ThreadPoolExecutor heavyExecutor = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS, heavyTaskQueue);

public static class SampleLightTask implements Runnable {

    @Override
    public void run() {
        System.out.println("I am " + this + " and running fast!");
    }

}

private static AtomicBoolean heavyTaskRunning = new AtomicBoolean();

public static class SampleHeavyTask implements Runnable {

    @Override
    public void run() {
        try {
            heavyTaskRunning.set(true);
            System.out.println("I am " + this + " and running quite slow!");
            final long start = System.currentTimeMillis();
            while (true) {
                //burn the CPU for ten senconds.
                if (System.currentTimeMillis()-start >= 10000L)
                    break;
            }
        } finally {
            heavyTaskRunning.set(false);;
        }
    }

}

public void shutDownNow() {
    this.lightExecutor.shutdownNow();
    this.heavyExecutor.shutdownNow();
}

public void runOrQueueLightTask(SampleLightTask lightOne) {
    this.lightExecutor.execute(lightOne);
}

public void runOrQueueHeavyTask(SampleHeavyTask heavyOne) {
    if (heavyTaskRunning.get()) {
        System.out.println("running, skipped new one: " + heavyOne);
        return;
    }

    this.heavyExecutor.execute(heavyOne);
}

public static void main(String[] args) throws Exception {
    TwoQueues q = new TwoQueues();

    final long start = System.currentTimeMillis();

    //Run the queues for 30 seconds, add CPU-light and CPU-weight tasks
    //every second.
    while (System.currentTimeMillis()-start<=30*1000L) {
        q.runOrQueueHeavyTask(new SampleHeavyTask());
        q.runOrQueueLightTask(new SampleLightTask());
        Thread.sleep(1000L);
    }

    q.shutDownNow();
}
}

以及运行输出：

I am toys.TwoQueues$SampleHeavyTask@6d0cecb2 and running quite slow!
I am toys.TwoQueues$SampleLightTask@6b87d20c and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@2ce07e6b
I am toys.TwoQueues$SampleLightTask@7fa0d111 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@16fdf48d
I am toys.TwoQueues$SampleLightTask@5fbd7d0e and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@115d533d
I am toys.TwoQueues$SampleLightTask@59c27402 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@6d4e5d57
I am toys.TwoQueues$SampleLightTask@33d232d1 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@79ec3264
I am toys.TwoQueues$SampleLightTask@1e081c5 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@3a67ad79
I am toys.TwoQueues$SampleLightTask@6cae00e3 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@13bc6ed3
I am toys.TwoQueues$SampleLightTask@380fe8c4 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@1c7ab89d
I am toys.TwoQueues$SampleLightTask@3cee5a06 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@44585f2a
I am toys.TwoQueues$SampleLightTask@5cfe174 and running fast!
I am toys.TwoQueues$SampleLightTask@12da89a7 and running fast!
I am toys.TwoQueues$SampleHeavyTask@49833c9c and running quite slow!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@47004b78
I am toys.TwoQueues$SampleLightTask@645ad7b2 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@8071a97
I am toys.TwoQueues$SampleLightTask@a62b39f and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@55fe910c
I am toys.TwoQueues$SampleLightTask@3be4d6ef and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@2cdb03a1
I am toys.TwoQueues$SampleLightTask@5ecb5608 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@777d57d6
I am toys.TwoQueues$SampleLightTask@4611dfe3 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@3f81d405
I am toys.TwoQueues$SampleLightTask@6486b4d5 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@47ca3f82
I am toys.TwoQueues$SampleLightTask@2f0f94a0 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@27e6ac83
I am toys.TwoQueues$SampleLightTask@1947e0ec and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@3dffb2eb
I am toys.TwoQueues$SampleLightTask@5e3b8219 and running fast!
I am toys.TwoQueues$SampleLightTask@14da67a4 and running fast!
I am toys.TwoQueues$SampleHeavyTask@eca4aae and running quite slow!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@2eced18
I am toys.TwoQueues$SampleLightTask@10c1c428 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@213526b0
I am toys.TwoQueues$SampleLightTask@287efdd8 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@294b84ad
I am toys.TwoQueues$SampleLightTask@1cf38f09 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@3a33a6b8
I am toys.TwoQueues$SampleLightTask@150697e2 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@63dd8136
I am toys.TwoQueues$SampleLightTask@634e3372 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@2313b44d
I am toys.TwoQueues$SampleLightTask@62a23d38 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@9615a1f
I am toys.TwoQueues$SampleLightTask@5663ae08 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@2a36bb87
I am toys.TwoQueues$SampleLightTask@6f51b1b7 and running fast!
running, skipped new one: toys.TwoQueues$SampleHeavyTask@5c6a9e79
I am toys.TwoQueues$SampleLightTask@5bca4955 and running fast!

///////////////////////////////////////////////////////////// //////////////////////

如果我正确理解您的要求，有两种类型的任务：A 类，CPU 密集型，以串行方式执行，并可能修改一些非线程安全的全局状态； B 类，不是 CPU 密集型的，需要尽快完成。

为什么不为此使用两个thread pools 和两个队列呢？这是一个完美的匹配。对于 A 类任务，将它们调度在最大并发设置为 1 的线程池中；对于类型 B，在另一个线程池中，最大并发设置为您的 CPU 核心/线程数或任何适合您需要的东西。这里甚至不需要“检查并优雅地失败”。

我以前自己写了很多这些原始的、低级的并发线程的东西，但是当线程池成为 JDK 中的标准库时，我再也不会回去了，无论是在服务器端（EE）还是客户端（这里是安卓）。设计简洁，性能好，代码少得多，当然，错误少得多。调试与并发相关的错误绝非易事。

【讨论】：

但是我不想让繁重的任务排队，我真的需要这样，当一个已经在运行时，跳过其余的并尽快继续。
好的，了解您的问题。这种设计只需稍作改动即可解决您的问题：如果有正在运行的繁重任务，则只需跳过排队即可。我个人还是比较喜欢这种设计的，因为比较简单，重轻任务的执行不会互相干扰，调试方便等。
如何做到“一点点改变”正是这个问题的问题！你不能说“就去做吧”；）
讲道理，因为程序员喜欢用代码说话 ;-)