Java如何使用ReentrantLock实现长轮询

时间:2022-02-06 23:00:17

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