Java如何使用ReentrantLock实现长轮询

Java代码

1. ReentrantLock

加锁阻塞，一个condition对应一个线程，以便于唤醒时使用该condition一定会唤醒该线程

/**
    * 获取探测点数据,长轮询实现
    * @param messageId
    * @return
    */
   public JSONObject getToutData(String messageId) {
       Message message = toutMessageCache.get(messageId);
       if (message == null) {
           // 等待
           lock.lock();
           try {
               Condition condition = lock.newCondition();
               conditionMap.put(messageId + "_data", condition);
               condition.await(CONNECTION_HOLD_TIMEOUT, TimeUnit.SECONDS); // 等待60s
           } catch (InterruptedException e) {
               // 等待超时， do nothing
           } finally {
               lock.unlock();
           }
       }

// 再次尝试获取
       message = toutMessageCache.get(messageId);
       if (message == null) {
           // 如果还没有， 返回空对象
           return null;
       }

byte[] bytes = message.getDataBytes();
       if (bytes == null) {
           return null;
       }
       String resStr = new String(bytes, StandardCharsets.UTF_8);
//        log.info("resStr: {}", resStr);
       JSONObject resObj;
       try {
           resObj = new JSONObject(resStr);
           resObj.put("invokeTime", DateUtil.format(new Date(resObj.getLong("invokeTime")), DatePattern.NORM_DATETIME_MS_PATTERN));
       } catch (Exception e) {
           resObj = new JSONObject();
       }

return resObj;
   }

2. 回调

当异步数据返回，使用上一步的condition唤醒线程

public void callback(Message message) {
   String messageId = message.getId();
   toutMessageCache.put(message.getId(), message);
   String messageDataId = messageId + "_data";
   if (conditionMap.containsKey(messageDataId)) {
       lock.lock();
       try {
           Condition condition = conditionMap.get(messageDataId);
           condition.signal();
       } catch (Exception e) {
           e.printStackTrace();
       } finally {
           lock.unlock();
           conditionMap.remove(messageDataId);
       }
   }
}

3. 唤醒

执行回调操作

public void distribute(Message message, ChannelHandlerContext ctx) {
  MessageType messageType = message.getMessageType();
  switch (messageType) {
      case TOUT_DATA_RESPONSE:
          // 数据响应
          toutService.callback(message);
          break;
  }

}

4. 调用

调用时，判断返回的值是否为空，如果为空，与前端约定，当返回该状态值时，应再次发起相同请求

/**
* 获取探测数据(使用长轮询实现)
* @param linkId
* @return
*/
@GetMapping("/data")
public ResultVO getToutData(String linkId) {
  JSONObject resObj = toutService.getToutData(linkId);
  if (resObj == null || resObj.isEmpty()) {
      return ResultVOUtil.error(ResultEnum.NO_MESSAGE_HOLD_CONNECTION);
  }
  return ResultVOUtil.success(resObj);
}

5.前端实现

简单使用递归实现了当数据返回无效时再次发起请求

let that = this
function getData() {
    if (toutStatus === statusEnum.start) {
        getToutData({
            linkId
        }).then(res => {
            if (res.code === ERROR_CODE_OK) {
                that.toutData = res.data
                toutStatus = statusEnum.resData
                that._btnStatus()
            } else {
                getData()
            }
        })
    }
}

// 递归循环调用
getData()

来源：https://blog.csdn.net/weixin_42096329/article/details/115470071