chws/user-service/src/main/java/com/mh/user/utils/CacheTools.java

package com.mh.user.utils;


import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author LJF
 * @version 1.0
 * @project TAD_Server
 * @description 缓存等待数据
 * @date 2023/7/4 08:45:16
 */
@Slf4j
public class CacheTools {

    /**
     * 响应消息缓存
     */
    private static final Cache<String, BlockingQueue<String>> responseMsgCache = CacheBuilder.newBuilder()
            .maximumSize(500)
            .expireAfterWrite(1000, TimeUnit.SECONDS)
            .build();


    /**
     * 等待响应消息
     *
     * @param key 消息唯一标识
     * @return ReceiveDdcMsgVo
     */
    public static String waitReceiveMsg(String key) {
        try {
            //设置超时时间
            String vo = Objects.requireNonNull(responseMsgCache.getIfPresent(key))
                    .poll(1000 * 5, TimeUnit.MILLISECONDS);

            //删除key
            responseMsgCache.invalidate(key);
            return vo;
        } catch (Exception e) {
            log.error("获取数据异常,sn={},msg=null", key);
            return "";
        }

    }

    /**
     * 初始化响应消息的队列
     *
     * @param key 消息唯一标识
     */
    public static void initReceiveMsg(String key) {
        responseMsgCache.put(key, new LinkedBlockingQueue<String>(1));
    }

    /**
     * 设置响应消息
     *
     * @param key 消息唯一标识
     */
    public static void setReceiveMsg(String key, String msg) {
        if (!responseMsgCache.asMap().containsKey(key)) {
            initReceiveMsg(key);
        }
        if (responseMsgCache.getIfPresent(key) != null) {
            responseMsgCache.getIfPresent(key).add(msg);
            return;
        }
        log.warn("sn {}不存在", key);
    }
}