Java代码
1. ReentrantLock
加锁阻塞，一个condition对应一个线程，以便于唤醒时使用该condition一定会唤醒该线程
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
/** * 获取探测点数据,长轮询实现 * @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唤醒线程
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
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. 唤醒
执行回调操作
|
1 2 3 4 5 6 7 8 9 10 |
public void distribute(Message message, ChannelHandlerContext ctx) { MessageType messageType = message.getMessageType(); switch (messageType) { case TOUT_DATA_RESPONSE: // 数据响应 toutService.callback(message); break ; }
} |
4. 调用
调用时，判断返回的值是否为空，如果为空，与前端约定，当返回该状态值时，应再次发起相同请求
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
/** * 获取探测数据(使用长轮询实现) * @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.前端实现
简单使用递归实现了当数据返回无效时再次发起请求
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
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() |
以上就是如何使用ReentrantLock实现长轮询的详细内容，更多关于ReentrantLock长轮询的资料请关注其它相关文章！
原文链接：https://blog.csdn.net/weixin_42096329/article/details/115470071
查看更多关于Java如何使用ReentrantLock实现长轮询的详细内容...