java - 如何从工作队列任务中获取结果？

Question

我实现了一个简单的工作队列，它从多个不同的线程接收任务。我希望这些任务向其源线程返回一个值，但不知道如何做到这一点。我考虑过使用 future，但没有办法明确设置 future 的值。我可以使用一个属性，但我不相信它们是线程安全的。

每个任务都是 DBRequest 的一个实现。实际内容有所不同，但所有 Activity 的结果都是一个字符串。异步线程创建 DBRequest 并将其提交到队列。队列运行任务，生成一个字符串。如何将该字符串返回到创建 DBRequest 的线程，以及如何使我的创建者线程等待结果？

public interface DBRequest {
    String execute(VdtsSysDB vdtsSysDB, BoardLoad currentLoad);
}

public class DBQueue implements Runnable {
    private static DBQueue dbQueue;
    private LinkedBlockingQueue<DBRequest> queue = new LinkedBlockingQueue<>();
    private VdtsSysDB vdtsSysDB = new VdtsSysDB();
    private ReentrantLock lock = new ReentrantLock();
    private static final Logger LOG = LoggerFactory.getLogger(DBQueue.class);
    private boolean kill = false;

    private BoardLoad currentLoad;
    private ProgressController progressController;

    public static DBQueue getInstance() {
        if (dbQueue == null) synchronized (DBQueue.class) {
            if (dbQueue == null)
                dbQueue = new DBQueue();
        }
        return dbQueue;
    }

    private DBQueue() {
    }

    public ReentrantLock getLock() {
        return lock;
    }

    @Override
    public void run() {
        LOG.info("Starting DBQueue loop. Kill {}.", kill);
        while (!kill) {
            DBRequest dbRequest = removeRequest();
            if (dbRequest != null) {
                lock.lock();
                String result = dbRequest.execute(vdtsSysDB, currentLoad);
                lock.unlock();
                if (progressController != null) Platform.runLater(() ->
                        progressController.updateDisplay(currentLoad));
            }
        }
        vdtsSysDB.getEntityManager().close();
    }


    public void addRequest(DBRequest dbRequest) {
        try {
            queue.add(dbRequest);
            LOG.info("Added request.");
        } catch (Exception e) {
            LOG.error("Can't add element.", e);
        }
    }

    private DBRequest removeRequest() {
        DBRequest result = null;
        try {
            //result = queue.poll(10, TimeUnit.SECONDS);
            result = queue.take();
        } catch (Exception e) {
            LOG.error("Exception.", e);
        }
        return result;
    }

    public void killDBQueue() {
        kill = true;
        LOG.info("Shutting down DBQueue.");
    }

    public static void start() {
        Thread thread = new Thread(DBQueue.getInstance(), "DBQueue Thread");
        thread.start();
        LOG.info("Starting DBQueue.");
    }

    public BoardLoad getCurrentLoad() {
        if (currentLoad == null)
            currentLoad = BoardLoad.getLastOpenLoad(vdtsSysDB);
        return currentLoad;
    }

    public void setCurrentLoad(BoardLoad proposedLoad) {
        // We can only have one open load, and by definition, the current load is open. So close it.
        if (this.currentLoad != null && !this.currentLoad.equals(proposedLoad)) {
            currentLoad.close(vdtsSysDB);
            if (proposedLoad != null) {
                this.currentLoad = vdtsSysDB.getEntityManager().find(BoardLoad.class, proposedLoad.getId());
            } else this.currentLoad = null;
        }
    }

    public ProgressController getProgressController() {
        return progressController;
    }

    public void setProgressController(ProgressController progressController) {
        this.progressController = progressController;
    }
}

编辑:我使用此队列来同步数据库访问，减少对锁的需求并确保请求按顺序完成。我不相信有任何其他方法可以实现这种异步请求 -> 同步请求更改。但我希望改变这种信念。

Answer 1

您应该在 DBRequest 接口(interface)中添加对提交线程的引用，并实现 setResult(String result) (或类似)方法来接收结果。您可以在提交线程 run() 方法上实现一个 CountDownLatch 等待(或类似)方法，以在将请求发送到队列并在 setResult< 中向下发送请求时等待设置闩锁 方法。如果我不清楚，请告诉我，我会详细说明。