JedisCluster性能指标的收集——Prometheus

所有的性能收集其实都是基于aop的，对于jedisCluster性能（jedis的一样）的收集，因为其并没有提供类似于mybatis的intercept机制，所以只能手动实现可供收集性能数据的切面。

说到切面，第一个就会想到代理，接下来就通过cglib实现对jedisCluster的代理。

动态代理的生成工厂SPI扩展：JedisClusterProxyFactory

@Spi
public interface JedisClusterProxyFactory {

    <T> T getProxy(Class<T> clz, String namespace, Class[] argumentTypes, Object[] arguments);
}

扩展实现：JedisClusterCglibProxyFactory

这里说一下cglib，其中的Enhancer在spring aop里使用很多，它相比于JDK的动态代理Proxy，可以为无接口的类创建代理。它会根据某个给定的类创建子类，并且所有非final的方法都带有回调钩子方法。

@SpiMeta(name = "cglib")
public class JedisClusterCglibProxyFactory implements JedisClusterProxyFactory {
    @Override
    @SuppressWarnings("unchecked")
    public <T> T getProxy(Class<T> clz, String namespace, Class[] argumentTypes, Object[] arguments) {
        //创建一个字节码增强器，用来创建代理类
        Enhancer enhancer = new Enhancer();
        //设置父类，即要被代理的Class对象
        enhancer.setSuperclass(clz);
        //设置回调的狗子方法
        enhancer.setCallback(new JedisClusterMethodInterceptor(namespace));
        return (T) enhancer.create(argumentTypes, arguments);
    }
}

其中的JedisClusterMethodInterceptor继承自cglib的方法拦截器MethodInterceptor的实现：从下面代码可以看出指标收集就是在拦截器中进行的。

public class JedisClusterMethodInterceptor extends BaseMethodInterceptor {
    public JedisClusterMethodInterceptor(String namespace) {
        this.namespace = namespace;
    }

    @Override
    protected Object innerInvoke(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
        return proxy.invokeSuper(obj, args);
    }

    @Override
    protected String getType() {
        return "jedisCluster";
    }
}

public abstract class BaseMethodInterceptor implements MethodInterceptor {
    private static Map<String, Stats> statsMap = new ConcurrentHashMap<>();
    String namespace;

    @Override
    public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
        //将请求的方法（如get/incr等），namespace标识使用caf框架中使用@EnableJedisClusterClient定义的namespace，如shua-kxct
        final String[] tags = {"method", method.getName(), "namespace", namespace};
        long begin = System.currentTimeMillis();
        Object result;
        //获取指标收集器
        Stats stats = getOrInitStats();
        //对每次redis请求gauge+1
        stats.incConc(tags);
        try {
            //通过代理调用jedisCluster具体的方法
            result = innerInvoke(obj, method, args, proxy);
        } catch (Throwable t) {
            //统计错误数
            stats.error(tags);
            throw t;
        } finally {
            //统计本次请求的执行时间
            stats.observe(System.currentTimeMillis() - begin, tags);
            //请求结束gauge-1
            stats.decConc(tags);
        }
        return result;
    }

    protected abstract Object innerInvoke(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable;

    protected abstract String getType();

    private Stats getOrInitStats() {
        String statsKey = getType() + "::" + namespace;
        Stats stats = statsMap.get(statsKey);
        if (stats != null) {
            return stats;
        }
        stats = Profiler.Builder.builder().type(getType()).namespace(namespace).build();
        statsMap.putIfAbsent(statsKey, stats);
        return statsMap.get(statsKey);
    }
}

这里有一个注意的地方，调用具体的jedisCluster方法是通过innerInvoke——>proxy.invokeSuper(obj, args)这种方式调用的，而不是像JDK的InvocationHandler通过method去调用具体方法，因为使用method调用会再次进入拦截器，所以只能通过第四个参数proxy调用。

动态代理生成工厂类：PjedisClusterFactory，用于生成jedisCluster代理对象。

public class PjedisClusterFactory {

    private static String fixNamespace() {
        String namespace = "default";
        if (StringUtils.isNotEmpty(JedisPropsHolder.NAMESPACE.get())) {
            namespace = JedisPropsHolder.NAMESPACE.get();
        }
        return namespace;
    }

    private static JedisCluster newJedisCluster(Class[] classes, Object[] objects) {
        //获取生成的代理对象
        final JedisCluster jedisCluster = ExtensionLoader.getExtensionLoader(JedisClusterProxyFactory.class)
                .getExtension("cglib")
                .getProxy(JedisCluster.class, fixNamespace(), classes, objects);
        //将代理对象注册到健康监测调度器，原理和之前介绍的Druid一样
        JedisClusterHealthTracker.addJedisCluster(fixNamespace(), jedisCluster);
        return jedisCluster;
    }
    private static Class[] CONST_1 = new Class[]{HostAndPort.class};
    public static JedisCluster newJedisCluster(HostAndPort node) {
        return newJedisCluster(CONST_1, new Object[]{node});
    }
    private static Class[] CONST_2 = new Class[]{HostAndPort.class, int.class};
    public static JedisCluster newJedisCluster(HostAndPort node, int timeout) {
        return newJedisCluster(CONST_2, new Object[]{node, timeout});
    }
    private static Class[] CONST_3 = new Class[]{HostAndPort.class, int.class, int.class};
    public static JedisCluster newJedisCluster(HostAndPort node, int timeout, int maxAttempts) {
        return newJedisCluster(CONST_3, new Object[]{node, timeout, maxAttempts});
    }
    private static Class[] CONST_4 = new Class[]{HostAndPort.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(HostAndPort node, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_4, new Object[]{node, poolConfig});
    }
    private static Class[] CONST_5 = new Class[]{HostAndPort.class, int.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(HostAndPort node, int timeout, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_5, new Object[]{node, timeout, poolConfig});
    }
    private static Class[] CONST_6 = new Class[]{HostAndPort.class, int.class, int.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(HostAndPort node, int timeout, int maxAttempts,
                                               final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_6, new Object[]{node, timeout, maxAttempts, poolConfig});
    }
    private static Class[] CONST_7 = new Class[]{HostAndPort.class, int.class, int.class, int.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(HostAndPort node, int connectionTimeout, int soTimeout,
                                               int maxAttempts, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_7, new Object[]{node, connectionTimeout, soTimeout, maxAttempts, poolConfig});
    }
    private static Class[] CONST_8 = new Class[]{HostAndPort.class, int.class, int.class, int.class, String.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(HostAndPort node, int connectionTimeout, int soTimeout,
                                               int maxAttempts, String password, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_8, new Object[]{node, connectionTimeout, soTimeout, maxAttempts, password, poolConfig});
    }
    private static Class aClass = new TypeToken<Set<HostAndPort>>() {}.getRawType();
    private static Class[] CONST_9 = new Class[]{aClass};
    public static JedisCluster newJedisCluster(Set<HostAndPort> nodes) {
        return newJedisCluster(CONST_9, new Object[]{nodes});
    }
    private static Class[] CONST_10 = new Class[]{aClass, int.class};
    public static JedisCluster newJedisCluster(Set<HostAndPort> nodes, int timeout) {
        return newJedisCluster(CONST_10, new Object[]{nodes, timeout});
    }
    private static Class[] CONST_11 = new Class[]{aClass, int.class, int.class};
    public static JedisCluster newJedisCluster(Set<HostAndPort> nodes, int timeout, int maxAttempts) {
        return newJedisCluster(CONST_11, new Object[]{nodes, timeout, maxAttempts});
    }
    private static Class[] CONST_12 = new Class[]{aClass, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(Set<HostAndPort> nodes, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_12, new Object[]{nodes, poolConfig});
    }
    private static Class[] CONST_13 = new Class[]{aClass, int.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(Set<HostAndPort> nodes, int timeout, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_13, new Object[]{nodes, timeout, poolConfig});
    }
    private static Class[] CONST_14 = new Class[]{aClass, int.class, int.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(Set<HostAndPort> jedisClusterNode, int timeout, int maxAttempts,
                                               final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_14, new Object[]{jedisClusterNode, timeout, maxAttempts, poolConfig});
    }
    private static Class[] CONST_15 = new Class[]{aClass, int.class, int.class, int.class, GenericObjectPoolConfig.class};
    public static JedisCluster newJedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,
                                               int maxAttempts, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_15, new Object[]{jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, poolConfig});
    }
    private static Class[] CONST_16 = new Class[]{aClass, int.class, int.class, int.class, String.class, GenericObjectPoolConfig.class};

    public static JedisCluster newJedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,
                                               int maxAttempts, String password, final GenericObjectPoolConfig poolConfig) {
        return newJedisCluster(CONST_16, new Object[]{jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, password, poolConfig});
    }
}

介绍完如何创建代理对象，然后在代理对象的钩子回调方法拦截器中收集指标，接下来看一下在项目中的实际应用。

封装的JedisClusterClient：其持有的JedisCluster实际是通过cglib生成的代理对象。

public JedisClusterClient(String namespace, JedisPoolConfig jedisPoolConfig, String address) throws NoSuchFieldException {
        this.namespace = namespace;
        this.address = address;
        this.jedisPoolConfig = jedisPoolConfig;
        String[] commonClusterRedisArray = address.split(",");
        Set<HostAndPort> jedisClusterNodes = new HashSet<>();
        for (String clusterHostAndPort : commonClusterRedisArray) {
            String host = clusterHostAndPort.split(":")[0].trim();
            int port = Integer.parseInt(clusterHostAndPort.split(":")[1].trim());
            jedisClusterNodes.add(new HostAndPort(host, port));
        }

        JedisPropsHolder.NAMESPACE.set(namespace);
        this.jedisCluster = PjedisClusterFactory.newJedisCluster(jedisClusterNodes, defaultConnectTimeout, defaultConnectMaxAttempts, jedisPoolConfig);

        Field field = BinaryJedisCluster.class.getDeclaredField("connectionHandler");
        field.setAccessible(true);
        connectionHandler = (JedisSlotBasedConnectionHandler) ReflectionUtils.getField(field, this.jedisCluster);
        Field cacheFiled = JedisClusterConnectionHandler.class.getDeclaredField("cache");
        cacheFiled.setAccessible(true);
        JedisClusterInfoCache cache = (JedisClusterInfoCache) ReflectionUtils.getField(cacheFiled, connectionHandler);
        assert cache != null;
        nodes = cache.getNodes();
    }

 最后是指标收集结果：