Лучший способ синхронизации асинхронной задачи

Question

Моя проблема: у меня есть беспроводная сеть 802.15.4, подключенная к последовательному порту (с использованием обертки). Я могу отправить пакеты в сеть и слушать входящие пакеты. Как вы можете себе представить, это очень асинхронно.

Здесь идет задача: я хочу отправить команды в сеть и дождаться ответа в одном вызове функции. Для примера: я хочу, чтобы получить температуру от узла с сетевым идентификатором 1338.

double getTemperature(int id) throws Exception { .... } 

Есть ли лучший способ ждать ответного сообщения другим, чем делать все это «синхронизирован (объект) ждать (..) уведомлять (..) «материал?

С наилучшими пожеланиями, bigbohne

Может быть, добавить специи:

Это должно кончиться в webinterface, где пользователь может запросить эти команды (либо через AJAX или напрямую). Я также думал о кэшировании значений ответа в базе данных. Но для некоторых команд у вас MUSS есть прямой ответ на успех/сбой.

Answer 1

Вы можете сделать это с помощью BlockingQueue, поэтому вам не придется вручную управлять какой-либо синхронизацией. Запрос может отправить сообщение, а затем вызвать take() в BlockingQueue, который будет ждать появления элемента. Элемент - это ответ, который вставляется в очередь любым слушателем, который у вас есть на порт при возврате ответа.

Вам просто нужно выполнить координацию, чтобы слушатель и реквестер имели одну и ту же очередь. Для поддержки этого сценария вам потребуется только одна очередь для каждого запроса.

// Send method 
BlockingQueue<Reply> q = new ArrayBlockingQueue<Reply>(1); 
serialCom.registerQueue(myRequest.getId()); 
serialCom.sendRequest(myRequest); 
return q.take(); 

//Listener 
BlockingQueue<Reply> q = queueMap.get(incomingMessage.getId()); 
q.add(incomingMessage.getReply()); 

Answer 2

Лучше всего обернуть его в некоторый многоразовый класс запроса/ответа. Таким образом, вы можете иметь стандартный способ запроса последовательного порта и ждать ответа. Однако этот класс, несомненно, должен будет использовать синхронизированное ключевое слово и ждать и уведомлять команды где-то в своей реализации. Это ключевая часть написания Java, и важно понять, как они работают.

Если вы должны были создать такую ​​функцию, как вы говорите, вы должны убедиться, что вы вызываете ее из потока, который может блокировать ожидание результата. Функция должна будет выполнить какой-то цикл ожидания внутри него, ожидая ответа от последовательного порта. Поэтому важно, чтобы всякий поток, на котором он работал, мог подождать.

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

Здесь есть несколько сложных проблем, и важно иметь хороший дизайн на месте, который удовлетворяет их все разумным образом.

Answer 3

Я не уверен, что я правильно понимаю вопрос, но я обычно использую Future<T> для этих задач.

Внутренняя реализация Future<T> использует ожидание и уведомление и все это, но сам интерфейс становится довольно чистым.

Future<Double> getTemperature(int id); 

В коде сообщения вы можете отображать входящие сообщения в отношении невыполненных фьючерсов. При заказе гарантированно вы могли бы сделать что-то вроде

class Something { 
    Map<Integer, Queue<Object>> requests; 

    synchronized Future<?> request(int id, Object data) { 
     MyFutureImpl future = new MyFuture(); 
     requests.get(id).add(future); 
     serializeAndSend(id, data); 
     return future; 
    } 

    void serializeAndSend(id, data) {...} 

    synchronized void response(int id, Object data) { 
     MyFutureImpl future = requests.get(id).remove(); 
     future.setValue(data); // This would fulfill the future, and any 
           // threads waiting in a get() will get the 
           // value and continue. 
    } 
} 

Где MyFutureImpl является очень простой в будущем-реализации. Я предполагаю, что есть поток связи, который вызывает response() при получении пакета.Я также предполагаю, что функция serializeAndSend() обрабатывает запись клиенту или блокирует до тех пор, пока операция записи не будет выполнена или не передана потоку связи.

Использование карт с параллельным доступом и структуры очереди может сделать ненужным synchronization. Если есть только один непогашенный вызов на один идентификатор, очередь становится ненужной, конечно.