如何从工作队列任务中获得结果

Question

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

每个任务都是DBRequest的一个实现。实际内容各不相同，但所有活动的结果都是一个字符串。异步线程创建一个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;
    }
}

EDIT：我使用这个队列来同步数据库访问，减少了对锁的需要，并确保请求按顺序完成。我不认为有任何其他方法可以实现这种异步请求->同步请求更改。但我很想改变这种信念。

Answer 1

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