跳过项目的 Java 优先级队列首先再次重新处理答案

【问题标题】：Java priority queues with skipped items reprocessed again first跳过项目的 Java 优先级队列首先再次重新处理
【发布时间】：2017-08-10 07:41:10
【问题描述】：

假设我们有一个JobItem，它有两个字段jobId 和dbTableName。我们还有一个 4 的 Executor 线程池。

最初，执行器将运行队列中的第一个作业项。如果后续队列头有相同的dbTableNamesay tableA，我想获取下一个队列项下一个不具有相同dbTableName的作业项并先执行。

由于第一个作业可能需要很长时间，我们最终可能会在再次处理 tableA 之前使用其他表处理多个其他作业项

我们要确保 tableA 的所有作业都按顺序处理。

我有另一个 List 保存当前正在运行的作业列表。

目前，我看到只有迭代队列项并检查当前正在运行的作业列表才能提供这样的功能。

有没有更好的方法来实现这一点？

谢谢

【问题讨论】：

您可以使用 FIFO 结构，并在同名表上已经运行作业时将作业推回队列中

标签： java multithreading synchronization

【解决方案1】：

您可以创建线程池和具有队列的工作线程。

public void execute(Runnable command) {

        final int key= command.getKey();
         //Some code to check if it is runing
        final int index = key != Integer.MIN_VALUE ? Math.abs(key) % size : 0;
        workers[index].execute(command);
    }

工人代码

 private final AtomicBoolean scheduled = new AtomicBoolean(false);

    private final BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>(maximumQueueSize);

    public void execute(Runnable command) {
        long timeout = 0;
        TimeUnit timeUnit = TimeUnit.SECONDS;
        if (command instanceof TimeoutRunnable) {
            TimeoutRunnable timeoutRunnable = ((TimeoutRunnable) command);
            timeout = timeoutRunnable.getTimeout();
            timeUnit = timeoutRunnable.getTimeUnit();
        }

        boolean offered;
        try {
            if (timeout == 0) {
                offered = workQueue.offer(command);
            } else {
                offered = workQueue.offer(command, timeout, timeUnit);
            }
        } catch (InterruptedException e) {
            throw new RejectedExecutionException("Thread is interrupted while offering work");
        }

        if (!offered) {
            throw new RejectedExecutionException("Worker queue is full!");
        }

        schedule();
    }

    private void schedule() {
        //if it is already scheduled, we don't need to schedule it again.
        if (scheduled.get()) {
            return;
        }

        if (!workQueue.isEmpty() && scheduled.compareAndSet(false, true)) {
            try {
                executor.execute(this);
            } catch (RejectedExecutionException e) {
                scheduled.set(false);
                throw e;
            }
        }
    }

    public void run() {
        try {
            Runnable r;
            do {
                r = workQueue.poll();
                if (r != null) {
                    r.run();
                }
            }
            while (r != null);
        } finally {
            scheduled.set(false);
            schedule();
        }
    }

【讨论】：

【解决方案2】：

对于每个表，您需要一个带有单独的作业输入队列的串行执行器。 Executor 从队列中获取作业并按顺序运行它们。串行执行器可以通过两种方式实现：作为线程或作为参与者。线程实现更简单，但需要更多内存。 Actor 实现需要对线程库的额外依赖。在您的情况下，actor 库可以像 SimpleActor.java 一样简单。

【讨论】：

感谢您提供单独的作业输入队列的想法。使用这种方法会更容易实现。