Я реализовал простую рабочую очередь, которая получает задачи из разных потоков. Я хочу, чтобы эти задачи возвращали значение их исходному потоку, но не могу понять, как это сделать.
Я подумал об использовании будущего, но нет способа явно установить значение будущего. Я мог бы использовать свойство, но я не верю, что оно безопасно для потоков.
Каждая задача - это реализация 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;
}
}
РЕДАКТИРОВАТЬ: я использую эту очередь для синхронизации доступа к базе данных, уменьшая потребность в блокировках и обеспечивая последовательное выполнение запросов. Я не верю, что есть какой-либо другой способ добиться такого рода асинхронного запроса -> синхронное изменение запроса.
Но я бы хотел изменить это убеждение.