如何处理 Java“作业”同步？

Question

我们有一组行动或“工作”，我们希望一次发生一个（而不是同时发生）。即：作业 A 不能在 B 发生时发生，并且不能同时运行两个 C 作业。

如果线程尝试同时运行作业，它们应该会收到错误。我们不应该只是对请求进行排队。

有些作业在用户发出请求时异步发生，我们返回状态消息和 ID，然后在服务器上异步处理作业。

我们正在寻求有关如何处理这种情况的建议。

一种选择是锁定共享对象：

public class Global {
  public static final Object lock = new Object();
}

public class JobA {
    public void go() {
        synchronized(Global.lock) {
            //Do A stuff
        }
    }
}

public class JobB {
    public void go() {
        synchronized(Global.lock) {
            //Do B stuff
        }
    }
}

这样做的问题：

1. 我们会将并发请求排队并且不返回错误消息（这是我们想要做的）
2. 如果JobA想要将消息写入队列以异步处理，我们如何保证当 JobA 中的另一个方法从队列中读取消息时，该方法将能够在 JobA< 的另一个实例之前获取 Global.lock 锁/代码> 开始？

关于更好的方法有什么建议吗？

Answer 1

您是否考虑过使用 单线程执行器 ?顾名思义，它使用单个线程来执行 Callables（类似于 Runnables - 请参阅文档），从而满足您对同步处理的要求。使用可用的 java.util.concurrent 类将极大地简化您的线程代码。

要处理在另一个作业正在运行时提交作业并生成错误的情况，请设置有界队列和拒绝处理程序。请参阅此页面和第 8.3.2/8.3.3 节特别是有关 ThreadPoolExecutor.setRejectedExecutionHandler()。

Answer 2

听起来您需要的是一个独占锁，您可以在尝试获取锁时避免阻塞。

幸运的是，Lock 接口有一个 tryLock 方法，因此您可以执行以下操作。

final Lock lock = new ReentrantLock();

....

void attemptJob( Runnable runnable )
{
    if (!lock.tryLock())
        throw new ConcurrentJobException();

    try
    {
       runnable.run();
    }
    finally
    {
       lock.unlock();
    }
}

如果你想运行异步，那么你应该将 Runnable（或者 Callable，如果你愿意）委托给 ExecutorService，并在它完成时释放锁。

编辑：例如（加上一些最终声明）

final Lock lock = new ReentrantLock();
final ExecutorService service = Executors.newSingleThreadExecutor();

....

void attemptJob( final Runnable runnable )
{
    if (!lock.tryLock())
        throw new ConcurrentJobException();

    service.execute( new Runnable()
    {
        public void run()
        {
            try
            {
                runnable.run();
            }
            finally
            {
                lock.unlock();
            } 
        }
    });
}

完成后不要忘记 shutdown() ExecutorService。

Answer 3

看看 Spring Batch。我认为它可以帮助你。

Answer 4

另一种可能性是使用已配置的 ThreadPoolExecutor。我相信这会起作用，但强烈鼓励读者亲自测试。

Executor executor = new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>());

上面的内容大约是 newSingleThreadExecutor 在 Executors 类中所做的事情。

为了满足“并发任务错误”的要求，我认为您可以将队列实现更改为 SynchronousQueue。

Executor executor = new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new SynchronousQueue<Runnable>());

这应该允许一个任务运行，如果在第一个任务完成之前提交第二个任务，则将调用 ThreadPoolExecutor RejectedExecutionHandler，默认情况下会抛出 RejectedExecutionException。