Java节流的 "消费者 "线程

Question

我想建立一个任务队列，每隔一段时间才运行一次。throttle 毫秒，而队列中还有任务。另外，由于任务正在进行在线查询，我需要在不同的线程中运行它们，以避免拖慢应用程序的其他部分。我得到的数据并不是 "高优先级 "的，我不需要快速运行队列，就像我想把请求速度降到涓涓细流一样。我尝试了下面的代码，尽管我几乎肯定它不会工作。

public class QueryManager implements Runnable {
    private Plugin p;
    private ScheduledExecutorService executor;
    private ConcurrentLinkedQueue<QueryRequest> jobs;

    public QueryManager(Plugin p) {
        this.p = p;
        this.executor = Executors.newSingleThreadScheduledExecutor();
        this.jobs = new ConcurrentLinkedQueue<>();
        this.executor.schedule(
                this,
                p.getConfig().getInt("requestThrottle", 250), //the int is used as default
                TimeUnit.MILLISECONDS);
    }

    public void addJob(QueryRequest req) {
        this.jobs.add(req);
        this.notify();
    }

    @Override
    public void run() {
        QueryRequest req = this.jobs.poll();
        if (req == null) {
            try {
                this.wait();
            } catch (Exception e) { e.printStackTrace(); }
        }
        else {
            req.run();
        }
    }
}

如我所料，结果是 IllegalMonitorStateException. 我对多线程没有经验，也不知道如何在这里尝试我的目标。(Consumer在标题中用了引号，因为它的行为有点像，但我不确定它在技术上是否是一个消费者。)

编辑: 通过删除 wait() 和 notify() 部分，我能够让它发挥作用，但它不符合我的性能偏好，即当队列中没有项目时暂停。

    //...snip...
    public void addJob(QueryRequest req) {
        this.jobs.add(req);
    }

    @Override
    public void run() {
        QueryRequest req = this.jobs.poll();
        if (req != null) {
            req.run();
        }
    }
}

Answer 1

你是什么 可能 真正需要的是 隔板但如果要自己做，你只需要修复你的。run 方法，如果没有请求等待，则立即返回。执行者的工作是调用 run 按计划反复进行。

需要注意的是，如果你管理的是自己的被执行人，你应该有一个? shutdown() 方法，将其干净地关闭；在容器环境中，该方法应该被注释为 @PreDestroy.

Answer 2

下面的代码没有错误，但只会执行列表中的一个作业。

public class QueryManager implements Runnable {
    private Plugin p;
    private ScheduledExecutorService executor;
    private ConcurrentLinkedQueue<QueryRequest> jobs;

    public QueryManager(Plugin p) {
        this.p = p;
        this.executor = Executors.newSingleThreadScheduledExecutor();
        this.jobs = new ConcurrentLinkedQueue<>();
        int throttle = Math.max(p.getConfig().getInt("requestThrottle", 250), 200);
        this.executor.schedule(this, throttle , TimeUnit.MILLISECONDS);
        Verbose.debug("Request throttle set to ", Integer.toString(throttle), TimeUnit.MILLISECONDS.name());
    }

    public void addJob(QueryRequest req) {
        this.jobs.add(req);
        Verbose.debug("Added query task to queue: ", req.getName());
    }

    @Override
    public void run() {
        QueryRequest req = this.jobs.poll();
        if (req != null) {
            Verbose.debug("Processing job...");
            req.run();
        }
    }
}

它仍然无法运行，但是现在它有一个完全不同的问题，我认为它最适合作为一个完全不同的问题。为了保持某种连续性，我将在该问题发布后编辑一个链接。编辑：新问题 Java "节流消费者 "只运行一次。