redis接口的java封裝

最近的項目需要使用redis。
主要需要實現(xiàn)幾個功能：
1 數(shù)據(jù)獲取
2 數(shù)據(jù)緩存
3 分布式鎖
4 帶鎖的數(shù)據(jù)處理

為了實現(xiàn)這幾個功能，對redis做了一層相對很薄的封裝，過程如下：

1 首先，我們需要定義redis的訪問接口，如下：

//數(shù)據(jù)查詢,更新接口
public interface DataAccess<T> {
    void set(String key, T value);

    void set(String key, T value, int validTime);

    T get(String key);

    List<T> popAll(String key);

    void push(String key, T value);

    //validTime = -1 表示無失效時間
    boolean setNX(String key, T value, int validTime);

    void delete(String key);

    boolean tryLock(String key, T value, int timeout);

    void unLock(String key);
}

接口的實現(xiàn)我們基于構(gòu)建匿名類的方式，數(shù)據(jù)訪問基于RedisTemplate, 代碼如下：

public class RedisAccessBuilder<T> {
    private RedisTemplate<String, T> redisTemplate;

    public RedisAccessBuilder(RedisTemplate<String, T> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    //1 設(shè)置json序列化方法
    //2 設(shè)置redis訪問器
    //3 構(gòu)造匿名類
    public DataAccess<T> buildDataAccess(Class<T> clazz, final DbDataGet<T> dbDataGet) {
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        //1
        if (clazz.equals(String.class)) {
            redisTemplate.setValueSerializer(new StringRedisSerializer());
        } else {
            redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<T>(clazz));
        }

        //2
        final ValueOperations<String, T> valueOperations= redisTemplate.opsForValue();
        final ListOperations<String, T> listOperations = redisTemplate.opsForList();

        //3
        return new DataAccess<T>() {
            @Override
            public void set(String key, T value) {
                valueOperations.set(key, value);
            }

            @Override
            public void set(String key, T value, int validTime) {
                set(key, value);
                redisTemplate.expire(key, validTime, TimeUnit.SECONDS);
            }

            @Override
            public T get(String key) {
                T data = valueOperations.get(key);

                //如果redis中沒有且設(shè)置了數(shù)據(jù)庫源,則從數(shù)據(jù)庫中獲取
                if (null == data
                        && null != dbDataGet) {
                    data = dbDataGet.get(key);

                    if (null != data) {
                        //默認保存一天
                        set(key, data, Constants.ONE_DAY);
                    }

                    try {
                        //避免數(shù)據(jù)庫壓力過大
                        Thread.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                return data;
            }

            @Override
            public List<T> popAll(String key) {
                String listKey = "list#" + key;
                List<T> resList = listOperations.range(listKey, 0, -1);

                delete(listKey);

                return resList;
            }

            @Override
            public void push(String key, T value) {
                String listKey = "list#" + key;
                listOperations.rightPush(listKey, value);
            }

            @Override
            public boolean setNX(String key, T value, int validTime) {
                if (valueOperations.setIfAbsent(key, value)) {
                    //todo throw exception
                    if (validTime > 0) {
                        return redisTemplate.expire(key, validTime, TimeUnit.SECONDS);
                    } else {
                        return true;
                    }
                }

                return false;
            }

            @Override
            public void delete(String key) {
                redisTemplate.delete(key);
            }

            @Override
            public boolean tryLock(String key, T value, int timeout) {
                String lockKey = "locked#" + key;
                long startTime = System.currentTimeMillis();
                int sleepTimes = 0;

                while (!setNX(lockKey, value, Constants.MAX_LOCK_TIME)) {
                    try {
                        Thread.sleep(1);

                        sleepTimes++;

                        if (sleepTimes % 100 == 0) {
                            //超時判斷
                            if (System.currentTimeMillis() - startTime > Constants.MAX_TIMEOUT * 1000) {
                                return false;
                            }
                        }
                    } catch (InterruptedException e) {
                        return false;
                    }
                }

                return true;
            }

            @Override
            public void unLock(String key) {
                String lockKey = "locked#" + key;

                delete(lockKey);
            }
        };
    }
}

在有了redis訪問接口之后，我們基于它，可以繼續(xù)定制redis緩存
2 redis緩存
我們的redis緩存分為兩種：
a 類似hashset，判斷一個key是否存在，實現(xiàn)如下：

//用redis實現(xiàn)的有時效緩存封裝
public class RedisCache {
    private DataAccess<String> dataAccess;

    public RedisCache(RedisTemplate<String, String> redisTemplate) {
        RedisAccessBuilder<String> redisAccessBuilder = new RedisAccessBuilder<>(redisTemplate);

        dataAccess = redisAccessBuilder.buildDataAccess(String.class, null);
    }

    //如果不存在,則插入并返回true
    //否則返回false
    public boolean cacheIfAbsent(String key, int validSecond) {
        String cacheKey = "cached#" + key;
        String cacheValue = "" + System.currentTimeMillis();

        return dataAccess.setNX(cacheKey, cacheValue, validSecond);
    }
}

b 類似hashtable，通過一個key，查詢value，為了保證數(shù)據(jù)的時效性，在實現(xiàn)中使用異步的方式從數(shù)據(jù)中查詢數(shù)據(jù)，保證當前數(shù)據(jù)的有效性，同時為了保證效率，使用一個基于內(nèi)存的hashmap來做第一層緩存，代碼如下：

public class MemCache<T> {
    private Map<String, T> map = new ConcurrentHashMap<String, T>();
    private DataAccess<T> dataAccess;
    private DbDataGet<T> dbDataGet;

    public MemCache(RedisTemplate<String, T> redisTemplate, Class<T> clazz, final DbDataGet<T> dbDataGet) {
        RedisAccessBuilder<T> redisAccessBuilder = new RedisAccessBuilder<>(redisTemplate);
        dataAccess = redisAccessBuilder.buildDataAccess(clazz, dbDataGet);
        this.dbDataGet = dbDataGet;

        asynUpdateData();
    }

    //1 檢查內(nèi)存
    //2 查詢redis(redis沒有則查詢數(shù)據(jù)庫)
    public T get(String key) {
        //1
        T value = map.get(key);

        if (null == value) {
            //2
            value = dataAccess.get(key);
            map.put(key, value);
        }

        return value;
    }

    //異步更新數(shù)據(jù),主要針對數(shù)據(jù)庫中的數(shù)據(jù)被修改,造成和redis及內(nèi)存不一致的情況
    //1 創(chuàng)建新的map
    //2 從原map中遍歷數(shù)據(jù)
    //3 從數(shù)據(jù)庫中獲取數(shù)據(jù),判斷是否和內(nèi)存中一致
    //4 如果不一致,則更新內(nèi)存及redis中的數(shù)據(jù)
    //5 將新的map作為當前的map
    private void asynUpdateData() {
        ExecutorService service = Executors.newFixedThreadPool(1);

        service.execute(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    int times = 0;

                    try {
                        //1
                        Map<String, T> newMap = new ConcurrentHashMap<>();

                        //2
                        for (String key : map.keySet()) {
                            times++;

                            //3
                            if (null != dataAccess.get(key)) {
                                T value = dbDataGet.get(key);

                                //4
                                if (value != map.get(key)) {
                                    newMap.put(key, value);
                                    dataAccess.set(key, value, Constants.ONE_DAY);
                                }
                            }

                            //每100條控制一下時間,避免數(shù)據(jù)庫壓力過大
                            if (times % 100 == 0) {
                                Thread.sleep(100);
                            }
                        }

                        //5
                        map = newMap;

                        Thread.sleep(Constants.FIVE_MINUTES);
                    } catch (InterruptedException e) {
                        log.info("interrupted " + e);
                    } catch (Exception e) {
                        log.info("exception " + e);
                    }
                }
            }
        });
    }
}

3&4 數(shù)據(jù)處理和分布式鎖我們放在一起處理，首先定義數(shù)據(jù)處理接口

//數(shù)據(jù)處理接口
public interface DataProcess<T> {
    DetailRes process(String key, T data);
}

之后定義帶鎖的數(shù)據(jù)處理接口

//鎖住key后執(zhí)行數(shù)據(jù)處理
public interface DataProcessWithLock<T> {
    DetailRes execute(String key, T data);
}

我們需要用戶自己實現(xiàn)DataProcess接口，并在構(gòu)建DataProcessWithLock的時候傳入，構(gòu)建DataProcessWithLock的代碼如下：

//帶鎖的數(shù)據(jù)處理封裝
public class RedisDataProcessWithLockBuilder {
    private DataAccess<String> dataAccess;

    public RedisDataProcessWithLockBuilder(RedisTemplate<String, String> redisTemplate) {
        RedisAccessBuilder<String> redisAccessBuilder = new RedisAccessBuilder<>(redisTemplate);

        dataAccess = redisAccessBuilder.buildDataAccess(String.class, null);
    }

    public <T> DataProcessWithLock<T> buildDataProcessWithLock(final DataProcess<T> dataProcess) {
        return new DataProcessWithLock<T>() {
            @Override
            public DetailRes execute(String key, T data) {
                //value為當前時間,方便追查問題
                String lockValue = "" + System.currentTimeMillis();

                try {
                    if (dataAccess.tryLock(key, lockValue, Constants.MAX_LOCK_TIME)) {
                        return dataProcess.process(key, data);
                    } else {
                        return new DetailRes(false, key + " locked failed");
                    }
                } catch (Exception e) {
                    return new DetailRes(false, key + " " + e.toString());
                } finally {
                    dataAccess.unLock(key);
                }
            }
        };
    }
}

有了前面的各種準備之后，我們用一個例子做結(jié)。
首先定義一個model：

@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {
    long userId;
    String name;
}

在對這個model處理前，我們要按userId將其鎖住，之后的處理就是簡單的將這個user打印出來，代碼如下：

@Service
public class RedisExample {
    @Autowired
    RedisTemplate<String, String> stringRedisTemplate;

    @Autowired
    RedisTemplate<String, User> userRedisTemplate;

    private DataProcessWithLock<User> dataProcessWithLock;

    @PostConstruct
    public void init() {
        RedisDataProcessWithLockBuilder redisDataProcessWithLockBuilder =
                new RedisDataProcessWithLockBuilder(stringRedisTemplate);

        dataProcessWithLock = redisDataProcessWithLockBuilder.buildDataProcessWithLock(new DataProcess<User>() {
            //用戶自己的實現(xiàn)
            @Override
            public DetailRes process(String key, User user) {
                System.out.println(user);

                return new DetailRes(true, "");
            }
        });
    }

    public void process() {
        User user = new User(111, "littlersmall");
        String key = "" + user.getUserId();

        dataProcessWithLock.execute(key, user);
    }
}

github地址如下

https://github.com/littlersmall/redis-access

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容