SPARROW 框架redis客户端封装实践

redis 本身有客户端，先抛出来一个问题？为什么要对redis客户端进行二次封装？

大概在11年时侯，第一次接触redis,那时侯研究过redis的各种数据结构，直接拿redis的客户端jedis直接用。公司安排人要对jedis进行封装，当时就很不理解，为什么非要封装一次才可以？

后来自己写框架，意识到一些东西是需要封装的，比如连接的打开和释放，比如一些危险的方法，比如keys * 比如flushdb 等

后来形成了这样的代码结构

T execute(Executor executor, KEY key)throws CacheConnectionException {

ShardedJedis jedis =null;

try {

Long startTime = System.currentTimeMillis();

jedis =this.pool.getResource();.//连接的获取

T result = executor.execute(jedis);

this.pool.returnResource(jedis);//连接的释放

Long endTime = System.currentTimeMillis();

if(this.cacheMonitor!=null) {

this.cacheMonitor.monitor(startTime, endTime, key);

}

return result;

}catch (JedisConnectionException e) {

this.pool.returnBrokenResource(jedis);//出错时连接释放

logger.error(this.getInfo() + SYMBOL.COLON + e.getMessage());

throw new CacheConnectionException(e.getMessage());

}

}

业务层会出现这样的代码

@Override

public Long addToSet(final KEY key,final Object value)throws CacheConnectionException {

//抛出connection exception异常提示业务方捕获处理

return redisPool.execute(new Executor() {

@Override

        public Long execute(ShardedJedis jedis) {

return jedis.sadd(key.key(),value.toString());

}

},key);

}

对连接的打开和释放进行了封装，避免业务端忘关链接而导致连接超时。抛出声明式异常，提示业务端处理链接断开的场景。

从业务使用上来讲基本不会有什么问题

但这种结构存在几个问题

1. 业务端存在jedis代码，如果想换jedis客户端，成本很大

2. 如果业务端想换另一种no sql 成本一样很大。

3 监控，当业务足够复杂时，对key的监控，问题排查和热点隔离成为痛点，如果让业务端对每一个 redis的请求点都加代码监控的话，这个成本依然很大。

一般公司业务都是从单体应用到分布式应用,如果某台机器报超时，对业务的key的监控就比较困难,比如

KEY为USER:1 USER:2 ...USER.N 进行监控，如何识别这是一组KEY?

 

 

REDIS分布式架构演进

所以对KEY的规范和统一监控就成为痛点

意淫部分

---

KEY的规范 

模块.业务类型:key id

对应数据结构如下:

 

redis KEY的定义

 

REDIS  相关类图

REDIS 操作接口定义

---

package com.sparrow.cache;

import com.sparrow.constant.cache.KEY;

import com.sparrow.exception.CacheConnectionException;

import com.sparrow.support.Entity;

import java.util.List;

import java.util.Map;

/**

* @author harry

* @date 2018/1/18

*/

public interface CacheClient {

Map hashGetAll(KEYkey)throws CacheConnectionException;

Map hashGetAll(KEYkey,Class keyClazz, Class dataClazz)throws CacheConnectionException;

Long getHashSize(KEYkey)throws CacheConnectionException;

Long getSetSize(KEYkey)throws CacheConnectionException;

Long removeFromOrderSet(KEYkey, Long from, Long to)throws CacheConnectionException;

Double getScore(KEYkey, Object value)throws CacheConnectionException;

Long getIndexOfOrderSet(KEYkey, Object value)throws CacheConnectionException;

Map getAllWithScore(KEYkey, Class clazz)throws CacheConnectionException;

Long getListSize(KEYkey)throws CacheConnectionException;

String hashGet(KEYkey, String hashKey)throws CacheConnectionException;

T hashGet(KEYkey, String hashKey, Class clazz)throws CacheConnectionException;

Long hashSet(KEYkey, String hashKey, Object value)throws CacheConnectionException;

//order set

    Long getOrderSetSize(KEYkey)throws CacheConnectionException;

Long addToSet(KEYkey, Object value)throws CacheConnectionException;

Long addToSet(KEYkey, String... value)throws CacheConnectionException;

Integer addToSet(KEYkey, Iterable values)throws CacheConnectionException;

Long addToList(KEYkey, Object value)throws CacheConnectionException;

Long removeFromList(KEYkey, Object value)throws CacheConnectionException;

Long removeFromSet(KEYkey, Object value)throws CacheConnectionException;

Long addToOrderSet(KEYkey, Object value, Long score)throws CacheConnectionException;

Long removeFromOrderSet(KEYkey, Object value)throws CacheConnectionException;

Boolean existInSet(KEYkey, Object value)throws CacheConnectionException;

Long addToList(KEYkey, String... value)throws CacheConnectionException;

Integer addToList(KEYkey, Iterable values)throws CacheConnectionException;

Long expire(KEYkey, Integer expire)throws CacheConnectionException;

Long delete(KEYkey)throws CacheConnectionException;

Long expireAt(KEYkey, Long expire)throws CacheConnectionException;

String setExpire(KEYkey, Integer seconds, Object value)throws CacheConnectionException;

String set(KEYkey, Entity value)throws CacheConnectionException;

String set(KEYkey, Object value)throws CacheConnectionException;

String get(KEYkey)throws CacheConnectionException;

T get(KEYkey, Class clazz)throws CacheConnectionException;

List getAllOfList(KEYkey)throws CacheConnectionException;

List getAllOfList(KEYkey, Class clazz)throws CacheConnectionException;

Long setIfNotExist(KEYkey, Object value)throws CacheConnectionException;

Long append(KEYkey, Object value)throws CacheConnectionException;

Long decrease(KEYkey)throws CacheConnectionException;

Long decrease(KEYkey, Long count)throws CacheConnectionException;

Long increase(KEYkey, Long count)throws CacheConnectionException;

Long increase(KEYkey)throws CacheConnectionException;

boolean bit(KEYkey, Integer offset)throws CacheConnectionException;

}

业务方必须统一按KEY类型读写redis，类型保护，保证KEY的规范一致。

 

KEY的定义

---

/*

* Licensed to the Apache Software Foundation (ASF) under one or more

* contributor license agreements.  See the NOTICE file distributed with

* this work for additional information regarding copyright ownership.

* The ASF licenses this file to You under the Apache License, Version 2.0

* (the "License"); you may not use this file except in compliance with

* the License.  You may obtain a copy of the License at

*

*    http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

package com.sparrow.constant.cache;

import com.sparrow.constant.magic.SYMBOL;

import com.sparrow.core.Pair;

import com.sparrow.support.ModuleSupport;

import com.sparrow.utility.StringUtility;

import java.util.Arrays;

/**

* Created by harry on 2018/1/8.

*/

public class KEY {

private Stringbusiness;

private ObjectbusinessId;

private Stringmodule;

private KEY(){}

private KEY(Builder builder) {

this.business = builder.business.getKey();

this.module=builder.business.getModule();

if (builder.businessId !=null) {

this.businessId = StringUtility.join(Arrays.asList(builder.businessId), SYMBOL.DOT);

}

}

public static KEY parse(String key){

if(StringUtility.isNullOrEmpty(key)){

return null;

}

KEY k=new KEY();

Pair businessWithId=Pair.split(key,SYMBOL.COLON);

k.businessId=businessWithId.getSecond();

String[] businessArray=businessWithId.getFirst().split("\\.");

k.module=businessArray[0];

k.business=businessWithId.getFirst();

return k;

}

public String key() {

if (StringUtility.isNullOrEmpty(this.businessId)) {

return this.business;

}

return this.business + SYMBOL.COLON +this.businessId;

}

public String getBusiness() {

return business;

}

public String getModule() {

return module;

}

public static class Business {

private Stringmodule;

private Stringkey;

public Business(ModuleSupport module, String... business) {

this.module = module.name();

this.key =this.module;

if (business !=null && business.length >0) {

this.key += SYMBOL.DOT + StringUtility.join(Arrays.asList(business), SYMBOL.DOT);

}

}

public String getKey() {

return key;

}

public String getModule() {

return module;

}

}

public static class Builder {

private Businessbusiness;

private Object[]businessId;

public Builder business(Business business) {

this.business = business;

return this;

}

public Builder businessId(Object... businessId) {

this.businessId = businessId;

return this;

}

public KEY build() {

return new KEY(this);

}

}

}

 

 

业务方可通过实现CacheMonitor 接口，对KEY进行统一监控

---

/**

* Created by harry on 2018/1/25.

*/

public class SparrowCacheMonitorimplements CacheMonitor{

@Override

    public void monitor(Long startTime, Long endTime, KEY key) {

可以对module 或business维护对KEY进行监控

System.out.println("module-"+key.getModule()+" business.type-"+key.getBusiness()+" key-"+key.key()+" start.time-"+startTime+" end.time-"+endTime);

}

}

 

DEMO实例

/**

* @author by harry

*/

public class RedisTest {

public static void main(String[] args)throws CacheConnectionException {

Container container =new SparrowContainerImpl();

//定义模块，一个业务会存在多个模块

        ModuleSupport OD=new ModuleSupport() {

@Override

            public String code() {

return "01";

}

@Override

            public String name() {

return "OD";

}

};

//相同模块下会存在多个业务

        KEY.Business od=new KEY.Business(OD,"POOL");

container.init();

CacheClient client = container.getBean("cacheClient");

//相同业务下存在多个KEY

        KEY key =new KEY.Builder().business(od).businessId("BJS","CHI","HU").build();

client.set(key,"test");

KEY k2=KEY.parse("OD.POOL:BJS.CHI.HU");

System.out.println("key:"+k2.key()+",module:"+k2.getModule()+" business:"+k2.getBusiness());

}

}

 

运行结果：

容器初始化...

module-OD business.type-OD.POOL key-OD.POOL:BJS.CHI.HU start.time-1516877958682 end.time-1516877958714

key:OD.POOL:BJS.CHI.HU,module:OD business:OD.POOL

 

源码下载

https://github.com/sparrowzoo/sparrow