我用Mybatis的方式封装了OLAP查询！

背景

相信做数据平台的朋友对OLAP并不陌生，主流的OLAP引擎有Clickhouse，Impala，Starrocks…以及公司二开的OLAP平台，本次要说的OLAP属于最后一种。
最近在做一个BI项目，业务背景很简单，就是一个数据展示平台。后端是SpringBoot + Mybatis 。 其中有一个比较特殊的是，我们不直接连接数据库，而是向OLAP平台传一个SQL，然后以HTTP请求的形式，从OLAP获得查询的结果。
由于Mybatis不支持配置HTTP形式数据源，我们这边后端同学的做法是，假装是数据库查询，实际用到的地方通过SqlSessionFactory获取执行SQL，然后将其封装在HTTP请求里。 对OLAP返回的Content 解析KeyValues的JSON，最终获得结果。

这种实现方式有一个问题就是， 我们使用Dao + XML的目的只是为了一段SQL，并不能直观的知道一个DAO里面的方法在什么地方使用到了。（因为SqlSessionFacatory获取SQL需要的是DAO名称和Method名称，所以以前是通过包路径获取）

Before

Service类里面的使用就是这种形式：

public DemoServiceImpl implements DemoService{@Autowired    OlapQueryUtils olapQueryUtils;// OlapQueryUtils是负责HTTP请求的工具类public Map<String,Object> getOlapData(RequestParam param){Map<String,Object> result = new HashMap<>();JSONArray json = olapQueryUtils.query("com.xx.xx.DemoDao.selectList", param);// 解析json成自己List<T>List<T> list = JSONUtils.parse(json, List<T>.class);result.put(Constants.DATA, list );return result;}
}

这段代码的问题有两个：

• com.xx.xx.DemoDao.selectList 是HardCode，如果这个类被移动或者重命名，这段代码会报错
• 返回的数据都要从JSONArray开始解析，JSON转换操作充斥所有Service。

Dao文件

public interface DemoDao{String selectList(RequestParam param); // no usage
}

这段简短的Dao代码，同样也有问题：

• 这个Dao代码的方法签名没有意义,至少返回类型没有意义，因为都是HTTP统一的JSONArray;
• 而且更致命的一点是no usage. IDE无法识别出来，容易被误删。

After

先不说怎么去实现，怎么去解决问题，看一下封装之后的代码片段。
Service：

public DemoServiceImpl implements DemoService{@AutowiredDemoDao demoDao;public Map<String,Object> getOlapData(RequestParam param){Map<String,Object> result = new HashMap<>();result.put(Constants.DATA, demoDao.selectList(param) );return result;}
}

Dao

@OlapMapper
public interface DemoDao{List<T> selectList(RequestParam param); // 1 usage
}

How

这里的原理很简单，就是模仿Mybatis用动态代理技术把DemoDao的动态bean注册到Spring。
Spring动态代理有三个关键步骤：

• Registry: 注册bean，让DemoDao可以按需被注入到Service中
• Factory： bean工厂，生产bean
• Proxy: 动态代理，提供接口方法实际实现。

Registry

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.ResourceLoader;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.core.type.classreading.CachingMetadataReaderFactory;
import org.springframework.data.util.AnnotatedTypeScanner;public class OlapDaoRegistry implements BeanDefinitionRegistryPostProcessor, ResourceLoaderAware, ApplicationContextAware {private ApplicationContext applicationContext;private ResourcePatternResolver resourcePatternResolver;private CachingMetadataReaderFactory metadataReaderFactory;private ResourceLoader resourceLoader;@Overridepublic void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {Set<Class<?>> sets = getOlapMappers();for (Class<?> bean : sets) {BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(bean);GenericBeanDefinition beanDefinition = (GenericBeanDefinition) builder.getRawBeanDefinition();beanDefinition.getConstructorArgumentValues().addGenericArgumentValue(bean);// 使用我们定义出来OlapFactory来注册beanbeanDefinition.setBeanClass(OlapDaoFactory.class);beanDefinition.setAutowireMode(GenericBeanDefinition.AUTOWIRE_BY_TYPE);registry.registerBeanDefinition(bean.getSimpleName(), beanDefinition);}}// 注册带@olapMapper的DAO文件@SneakyThrowsprivate Set<Class<?>> getOlapMappers() {AnnotatedTypeScanner scanner = new AnnotatedTypeScanner(OlapMapper.class);return scanner.findTypes("com.xx.xx");}@Overridepublic void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {}@Overridepublic void setResourceLoader(ResourceLoader resourceLoader) {this.resourcePatternResolver = new PathMatchingResourcePatternResolver();this.metadataReaderFactory = new CachingMetadataReaderFactory(resourceLoader);this.resourceLoader = resourceLoader;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}}

Factory

import org.springframework.beans.factory.FactoryBean;import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;public class OlapDaoFactory<T> implements FactoryBean<T> {private final Class<T> clazz;public OlapDaoFactory(Class<T> clazz) {this.clazz = clazz;}@Override@SuppressWarnings({Constant.Suppress.UNCHECKED})public T getObject() {// 使用我们定义的OlapServiceProxy来代理需要提供的BeanInvocationHandler invocationHandler = new OlapServiceProxy<>(clazz);return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, invocationHandler);}@Overridepublic Class<?> getObjectType() {return clazz;}
}

Proxy

// 跟Mybatis一样支持数据源的动态切换，以Clickhouse和Starrocks两种为例// 这里通过moduleName来查看是否支持数据源，你也可以去掉这个设计// 因为缓存可以大幅度提高OLAP select的效率，这里引入了缓存的设计import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;import java.lang.annotation.Annotation;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;@Slf4j
@RequiredArgsConstructor
public class OlapServiceProxy<T> implements InvocationHandler {private final Class<T> clazz;private String getDaoPrefix() {return clazz.getName() + ".";}private String getRedisKeyPre() {String daoPrefix = getDaoPrefix();daoPrefix = daoPrefix.replace("com.xx.", "");if (!daoPrefix.startsWith("appName.")) {daoPrefix = "appName." + daoPrefix;}return daoPrefix.replace("\\.", ":");}private static void preCheck(String module) {if (!module.contains("-")) {throw new UnsupportedOperationException("模块名应该包含'-'");}}private String getMethodName(String methodName) {return getDaoPrefix() + methodName;}private JSONArray queryCkWithCache(Object request, String method, String module) {preCheck(module);CkModelUtils ckModelUtils = SpringReflectUtils.getBean(CkModelUtils.class);return ckModelUtils.getCacheOrOlapArrayResultData(request, getMethodName(method), getRedisKeyPre() + module, Map.class, module);}private JSONArray queryCk(Object request, String method, String module) {preCheck(module);CkModelUtils ckModelUtils = SpringReflectUtils.getBean(CkModelUtils.class);return ckModelUtils.getDataFromOlap(request, getMethodName(method));}private JSONArray querySrWithCache(Object request, String method, String module) {preCheck(module);SrModelUtils srModelUtils = SpringReflectUtils.getBean(SrModelUtils.class);return srModelUtils.getCacheOrOlapArrayResultData(request, getDaoPrefix(), method, getRedisKeyPre() + module, Map.class, module);}private JSONArray querySr(Object request, String method, String module) {preCheck(module);SrModelUtils srModelUtils = SpringReflectUtils.getBean(SrModelUtils.class);return srModelUtils.getModelData(request, getDaoPrefix(), method, module);}@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {// fail fastif (Object.class.equals(method.getDeclaringClass())) {log.info("invoke equals method ");return method.invoke(this, args);}Datasource datasource = getDatasource(method);Object request = wrapParam(method, args);JSONArray data = queryFromOlap(method, request, datasource);return processReturnData(method, data);}/*** 从Olap查询获取JSONArray返回数据* @param method 被代理的方法* @param request 请求对象* @param datasource 数据源, 目前可选: CK,SR* @return olap返回的keyValues JSONArray*/private JSONArray queryFromOlap(Method method, Object request, Datasource datasource) {String module = "通用-动态代理";if (method.isAnnotationPresent(Module.class)) {module = method.getAnnotation(Module.class).value();}boolean isCache = this.clazz.isAnnotationPresent(Cache.class) || method.isAnnotationPresent(Cache.class);if (isCache) {if (datasource.equals(Datasource.CK)) {return queryCkWithCache(request, method.getName(), module);} else {return querySrWithCache(request, method.getName(), module);}} else {if (datasource.equals(Datasource.CK)) {return queryCk(request, method.getName(), module);} else {return querySr(request, method.getName(), module);}}}/*** 返回值处理* @param method 被代理的方法, 用来获取返回值类型* @param data olap查询到的JSONArray* @return 根据方法签名返回值,返回转换后的数据*/private @Nullable Object processReturnData(Method method, JSONArray data) {Class<?> returnType = method.getReturnType();// JSONArray直接返回if (returnType.getName().equals(JSONArray.class.getName())) {return data;}// 数组和列表-> SelectMany 就返回多行if (returnType.isArray() || Collection.class.isAssignableFrom(returnType)) {return data.toJavaObject(method.getGenericReturnType());} else {// 返回一行直接取第一个转成对象if (CollectionUtils.isEmpty(data)) return null;if (isNativeType(returnType)) {JSONObject jsonObject = data.getJSONObject(0);String key = jsonObject.keySet().iterator().next();return jsonObject.getObject(key, returnType);}return data.getObject(0, returnType);}}// 数据源: 默认CK -> 类注解覆盖 -> 方法注解覆盖private Datasource getDatasource(Method method) {Datasource datasource = Datasource.CK;if (this.clazz.isAnnotationPresent(DS.class)) {datasource = this.clazz.getAnnotation(DS.class).value();}if (method.isAnnotationPresent(DS.class)) {datasource = method.getAnnotation(DS.class).value();}return datasource;}private Object wrapParam(Method method, Object[] args) {if (args == null || args.length == 0) return null;if (args.length > 1) {Map<String, Object> paramMap = new HashMap<>();Annotation[][] annotations = method.getParameterAnnotations();for (int i = 0; i < args.length; i++) {Object arg = args[i];String key =Arrays.stream(annotations[i]).filter(x -> x instanceof Param).findFirst().map(x -> ((Param) x).value()).orElseThrow(UnsupportedOperationException::new);paramMap.put(key, arg);}return paramMap;} else {return args[0];}}/*** 判断是不是直接类型*/private boolean isNativeType(Class<?> clazz) {String clazzName = clazz.getName();Class<?>[] nativeClasses = {String.class, Integer.class, Boolean.class, Double.class, Long.class, Float.class, Short.class};return Arrays.stream(nativeClasses).anyMatch(x -> clazzName.equals(x.getName()));}
}

自定义注解

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Module {String value();
}@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DS {Datasource value();
}/*** OlapMapper注解* <p>*      - 用在整个Dao文件上表示所有的方法均走缓存* <p>*      - 用在某个具体方法上面修改该方法的缓存配置*/
@Documented
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface Cache {}@Component
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface OlapMapper {
}@Target({ElementType.PARAMETER})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Param {String value() ;
}

后记

这篇代码量比较大，就是说这个是一个用得着的时候可以直接抄的博客，一切是为了代码的可维护性！