Как я могу получить результаты от моей задачи очереди работы? - PullRequest
0 голосов
/ 13 ноября 2018

Я реализовал простую рабочую очередь, которая получает задачи из разных потоков. Я хочу, чтобы эти задачи возвращали значение их исходному потоку, но не могу понять, как это сделать. Я подумал об использовании будущего, но нет способа явно установить значение будущего. Я мог бы использовать свойство, но я не верю, что оно безопасно для потоков.

Каждая задача - это реализация 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;
    }
}

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

1 Ответ

0 голосов
/ 13 ноября 2018

Вы должны добавить ссылку на поток отправки в вашем интерфейсе DBRequest и реализовать метод setResult(String result) (или аналогичный) для получения результата.Вы можете внедрить CountDownLatch ожидание (или аналогичное) в вашем потоке подачи run() метод ожидания, когда настройка фиксируется при отправке запроса в очередь и вниз в методе setResult.Если мне не ясно, просто дайте мне знать, и я уточню.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...