這次開發功能是OEM統計報表。統計報表的數據由大數據平台部的同事收集，數據的展示由我們部門開發。
大數據那邊使用 Kylin 分布式分析引擎（kylin官方文檔）。
Kylin 雖比較偏向大數據相關，但最終大佬們決定把訪問 Kylin 的 Dubbo 接口也由我們開發，比較坑。
解決方案一：Mybatis
首先我們搭建一個 shop-mod-bigdata (+ shop-mod-bigdata-interface) 模塊用於訪問 Kylin，暴露 Dubbo 服務。
Kylin 官方文檔提供了 JDBC 的方式訪問Kylin。
為了長期考慮，是否可以用上連接池？
Kylin 有提供 JDBC 的方式訪問，那就應該會遵循 JDBC 的規范，
參考 Mysql 的連接池配置進行配置
<!-- Kylin 數據源（OEM） -->
<bean id="kylinOemDataSource" class="org.apache.tomcat.jdbc.pool.DataSource"
destroy-method="close">
<property name="driverClassName" value="${ds.kylin.jdbc.Driver}" />
<property name="url" value="${ds.kylin.oem.jdbc.url}" />
<property name="username" value="${ds.kylin.oem.jdbc.username}" />
<property name="password" value="${ds.kylin.oem.jdbc.password}" />
<!-- 連接池配置 -->
<property name="testWhileIdle" value="${ds.kylin.testWhileIdle}" />
<property name="testOnBorrow" value="${ds.kylin.testOnBorrow}" />
<property name="testOnReturn" value="${ds.kylin.testOnReturn}" />
<property name="validationQuery" value="${ds.kylin.validationQuery}" />
<property name="validationInterval" value="${ds.kylin.validationInterval}" />
<property name="timeBetweenEvictionRunsMillis" value="${ds.kylin.timeBetweenEvictionRunsMillis}" />
<property name="maxActive" value="${ds.kylin.maxActive}" />
<property name="maxIdle" value="${ds.kylin.maxIdle}" />
<property name="minIdle" value="${ds.kylin.minIdle}" />
<property name="maxWait" value="${ds.kylin.maxWait}" />
<property name="initialSize" value="${ds.kylin.initialSize}" />
<property name="removeAbandonedTimeout" value="${ds.kylin.removeAbandonedTimeout}" />
<property name="removeAbandoned" value="${ds.kylin.removeAbandoned}" />
<property name="logAbandoned" value="${ds.kylin.logAbandoned}" />
<property name="minEvictableIdleTimeMillis" value="${ds.kylin.minEvictableIdleTimeMillis}" />
<property name="jmxEnabled" value="${ds.kylin.jmxEnabled}" />
<property name="jdbcInterceptors" value="${ds.kylin.jdbcInterceptors}" />
</bean>
加上連接池配置后，下面就要驗證和測試了。
1：調試，跟源碼。
2：為便於觀察運行細節是否正常，修改 log4j 日志配置
log4j.logger.org.apache.tomcat.jdbc=DEBUG log4j.logger.org.springframework.jdbc=DEBUG log4j.logger.org.apache.kylin.jdbc=DEBUG log4j.logger.org.apache.ibatis=DEBUG log4j.logger.org.mybatis.spring=DEBUG
3：測試有沒有內存泄漏，長時間運行時是否有問題：CPU 是否異常，線程數是否異常，連接池是否異常，JVM垃圾回收是否異常，
剛好用用學到的 JVM 相關的東東，這里我就直接使用 JDK 自帶的工具（jvisualvm），
入口的話就直接用單元測試+線程池來跑
/**
* Kylin 訪問壓測
*
* @author zhangjl
* @date 2018年3月9日
*/
public class KylinPressureTest extends TestBase {
@Resource
private PersonFacade personFacade;
@Test
public void pressureTest() {
ExecutorService threadPool = Executors.newFixedThreadPool(10);
final AtomicInteger counter = new AtomicInteger();
final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
// 生成垃圾
Executors.newScheduledThreadPool(1).scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
int index = generateGarbage();
LOGGER.info("index={}", index);
}
}, 5, 5, TimeUnit.SECONDS);
// 訪問 Kylin
for (int i = 0; i < 10000; i++) {
LOGGER.info("第" + (i + 1) + "輪開始");
for (int j = 0; j < 10; j++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
int currNum = counter.addAndGet(1);
try {
PageInfo page = new PageInfo();
page.setPagination(true);
PersonDto condition = new PersonDto();
condition.setGender(counter.get());
ApiResult<PageResult<PersonDto>> result = personFacade.page(condition, page);
LOGGER.info("Kylin訪問結果, result={}", result);
} catch (Exception e) {
LOGGER.error("Kylin訪問異常", e);
}
LOGGER.info("Kylin訪問，當前為第{}個，time={}", currNum, sdf.format(new Date()));
}
});
}
try {
Thread.sleep(1L * 60L * 1000L);
} catch (InterruptedException e) {
LOGGER.error("線程中斷", e);
}
}
// 等待
waitingUserInputExit();
}
private void waitingUserInputExit() {
Scanner scanner = null;
try {
scanner = new Scanner(System.in);
while (true) {
String input = scanner.next();
if ("exit".equals(input)) {
break;
}
}
} catch (Exception e) {
LOGGER.error("輸出異常", e);
} finally {
if (null != scanner) {
scanner.close();
}
}
}
private int generateGarbage() {
int index = 0;
byte[] array1 = new byte[100 * 1024 * 1024]; // 100M
// byte[] array2 = new byte[100 * 1024 * 1024]; // 100M
array1[index++] = new Byte("1");
// array2[index++] = new Byte("2");
return index;
}
}
修改 VM arguments 參數：
-Xmx1024m -Xms1024m -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintHeapAtGC -Xloggc:mygc.log
結果圖：
根據測試結果，連接池應該沒什么大問題。
感覺應該可以用 Mybatis ？那就加上 Mybatis 的配置，再試一波。
簡單 SQL 驗證沒問題。
又遇到一個問題：分頁。
Mysql分頁使用的是：LIMIT ${pageSize}, ${offset}
Kylin分頁使用的是：LIMIT ${pageSize} OFFSET ${offset}
好像區別並不大，嗯嗯，完全可以基於以前的 Mysql Dialect 改代碼。
<!-- Kylin jdbcTemplate（OEM） --> <!-- <bean id="kylinOemJdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate"> <property name="dataSource" ref="kylinOemDataSource"></property> </bean> --> <!-- Kylin 不配置統一事務管理器（OEM） --> <!-- Kylin sqlSessionFactory（OEM） --> <bean id="kylinOemSqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean"> <property name="dataSource" ref="kylinOemDataSource" /> <property name="plugins"> <list> <bean class="yunnex.shop.bigdata.common.kylin.KylinMybatisLogInterceptor"></bean> <bean class="yunnex.shop.bigdata.common.kylin.KylinMybatisPaginationInterceptor"></bean> </list> </property> <property name="mapperLocations"> <array> <value>classpath*:/yunnex/shop/bigdata/oem/**/*.xml</value> </array> </property> </bean> <!-- Kylin MapperScannerConfigurer（OEM） --> <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer"> <property name="basePackage" value="yunnex.shop.bigdata.oem.**.mapper,yunnex.shop.bigdata.oem.**.dao" /> <property name="sqlSessionFactoryBeanName" value="kylinOemSqlSessionFactory"></property> </bean>
/**
* 分頁攔截器
*
* @author zhangjl
* @date 2018年3月9日
*/
@Intercepts({@Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class})})
public class KylinMybatisPaginationInterceptor implements Interceptor {
private Logger LOGGER = LoggerFactory.getLogger(KylinMybatisPaginationInterceptor.class);
Dialect dialect = new KylinDialect();
@Value("${sql.rows.max.return}")
private Integer rowlimit;
@SuppressWarnings("unchecked")
public Object intercept(Invocation invocation) throws Throwable {
MappedStatement mappedStatement = (MappedStatement) invocation.getArgs()[0];
Object parameter = invocation.getArgs()[1];
BoundSql boundSql = mappedStatement.getBoundSql(parameter);
String sqlId = mappedStatement.getId();
String originalSql = boundSql.getSql().trim();
RowBounds rowBounds = (RowBounds) invocation.getArgs()[2];
Object parameterObject = boundSql.getParameterObject();
if (boundSql.getSql() == null || "".equals(boundSql.getSql())) {
return null;
}
// 分頁參數--上下文傳參
PageInfo page = null;
// map傳參每次都將currentPage重置,先判讀map再判斷context
if (parameterObject instanceof PageInfo) {
page = (PageInfo) parameterObject;
} else if (parameterObject instanceof Map) {
Map<String, Object> map = (Map<String, Object>) parameterObject;
if (map.containsKey("page")) {
page = (PageInfo) map.get("page");
}
} else if (null != parameterObject) {
Field pageField = ReflectionUtil.getFieldByFieldName(parameterObject, "page");
if (pageField != null) {
page = (PageInfo) ReflectionUtil.getValueByFieldName(parameterObject, "page");
}
}
// 后面用到了context的東東
if (page != null && page.isPagination() == true) {
if (page.getPageSize() > rowlimit) {
LOGGER.warn("[toolarge_pagesize] page size greater than {},#sqlid:{}#,#pagesize:{}#,#sql:{}#", rowlimit, sqlId, page.getPageSize(),
originalSql);
page.setPageSize(rowlimit);
}
int totalRows = page.getTotalRows();
// 得到總記錄數
if (totalRows == 0 && page.isNeedCount()) {
StringBuilder countSql = new StringBuilder();
countSql.append(KylinPageHepler.getCountString(originalSql));
Connection connection = mappedStatement.getConfiguration().getEnvironment().getDataSource().getConnection();
PreparedStatement countStmt = connection.prepareStatement(countSql.toString());
BoundSql countBS = new BoundSql(mappedStatement.getConfiguration(), countSql.toString(), boundSql.getParameterMappings(),
parameterObject);
Field metaParamsField = ReflectionUtil.getFieldByFieldName(boundSql, "metaParameters");
if (metaParamsField != null) {
MetaObject mo = (MetaObject) ReflectionUtil.getValueByFieldName(boundSql, "metaParameters");
ReflectionUtil.setValueByFieldName(countBS, "metaParameters", mo);
}
setParameters(countStmt, mappedStatement, countBS, parameterObject);
ResultSet rs = countStmt.executeQuery();
if (rs.next()) {
totalRows = rs.getInt(1);
}
rs.close();
countStmt.close();
connection.close();
}
// 分頁計算
page.init(totalRows, page.getPageSize(), page.getCurrentPage());
if (rowBounds == null || rowBounds == RowBounds.DEFAULT) {
rowBounds = new RowBounds(page.getPageSize() * (page.getCurrentPage() - 1), page.getPageSize());
}
// 分頁查詢 本地化對象 修改數據庫注意修改實現
String pagesql = dialect.getLimitString(originalSql, rowBounds.getOffset(), rowBounds.getLimit());
invocation.getArgs()[2] = new RowBounds(RowBounds.NO_ROW_OFFSET, RowBounds.NO_ROW_LIMIT);
BoundSql newBoundSql = new BoundSql(mappedStatement.getConfiguration(), pagesql, boundSql.getParameterMappings(),
boundSql.getParameterObject());
Field metaParamsField = ReflectionUtil.getFieldByFieldName(boundSql, "metaParameters");
if (metaParamsField != null) {
MetaObject mo = (MetaObject) ReflectionUtil.getValueByFieldName(boundSql, "metaParameters");
ReflectionUtil.setValueByFieldName(newBoundSql, "metaParameters", mo);
}
MappedStatement newMs = copyFromMappedStatement(mappedStatement, new BoundSqlSqlSource(newBoundSql));
invocation.getArgs()[0] = newMs;
}
return invocation.proceed();
}
public static class BoundSqlSqlSource implements SqlSource {
BoundSql boundSql;
public BoundSqlSqlSource(BoundSql boundSql) {
this.boundSql = boundSql;
}
public BoundSql getBoundSql(Object parameterObject) {
return boundSql;
}
}
public Object plugin(Object arg0) {
return Plugin.wrap(arg0, this);
}
public void setProperties(Properties arg0) {
}
/**
* 對SQL參數(?)設值,參考org.apache.ibatis.executor.parameter.DefaultParameterHandler
*
* @param ps
* @param mappedStatement
* @param boundSql
* @param parameterObject
* @throws SQLException
*/
@SuppressWarnings({"rawtypes", "unchecked"})
private void setParameters(PreparedStatement ps, MappedStatement mappedStatement, BoundSql boundSql, Object parameterObject) throws SQLException {
ErrorContext.instance().activity("setting parameters").object(mappedStatement.getParameterMap().getId());
List<ParameterMapping> parameterMappings = boundSql.getParameterMappings();
if (parameterMappings != null) {
Configuration configuration = mappedStatement.getConfiguration();
TypeHandlerRegistry typeHandlerRegistry = configuration.getTypeHandlerRegistry();
MetaObject metaObject = parameterObject == null ? null : configuration.newMetaObject(parameterObject);
for (int i = 0; i < parameterMappings.size(); i++) {
ParameterMapping parameterMapping = parameterMappings.get(i);
if (parameterMapping.getMode() != ParameterMode.OUT) {
Object value;
String propertyName = parameterMapping.getProperty();
PropertyTokenizer prop = new PropertyTokenizer(propertyName);
if (parameterObject == null) {
value = null;
} else if (typeHandlerRegistry.hasTypeHandler(parameterObject.getClass())) {
value = parameterObject;
} else if (boundSql.hasAdditionalParameter(propertyName)) {
value = boundSql.getAdditionalParameter(propertyName);
} else if (propertyName.startsWith(ForEachSqlNode.ITEM_PREFIX) && boundSql.hasAdditionalParameter(prop.getName())) {
value = boundSql.getAdditionalParameter(prop.getName());
if (value != null) {
value = configuration.newMetaObject(value).getValue(propertyName.substring(prop.getName().length()));
}
} else {
value = metaObject == null ? null : metaObject.getValue(propertyName);
}
TypeHandler typeHandler = parameterMapping.getTypeHandler();
if (typeHandler == null) {
throw new ExecutorException(
"There was no TypeHandler found for parameter " + propertyName + " of statement " + mappedStatement.getId());
}
typeHandler.setParameter(ps, i + 1, value, parameterMapping.getJdbcType());
}
}
}
}
private MappedStatement copyFromMappedStatement(MappedStatement ms, SqlSource newSqlSource) {
Builder builder = new MappedStatement.Builder(ms.getConfiguration(), ms.getId(), newSqlSource, ms.getSqlCommandType());
builder.resource(ms.getResource());
builder.fetchSize(ms.getFetchSize());
builder.statementType(ms.getStatementType());
builder.keyGenerator(ms.getKeyGenerator());
builder.keyProperty(buildKeyProperty(ms.getKeyProperties()));
builder.timeout(ms.getTimeout());
builder.parameterMap(ms.getParameterMap());
builder.resultMaps(ms.getResultMaps());
builder.useCache(ms.isUseCache());
builder.cache(ms.getCache());
MappedStatement newMs = builder.build();
return newMs;
}
private static String buildKeyProperty(String[] props) {
if (null != props && props.length > 0) {
StringBuilder sb = new StringBuilder();
for (String p : props) {
sb.append(p).append(",");
}
return sb.substring(0, sb.length() - 1);
}
return null;
}
}
/**
* Kylin Dialect
*
* @author zhangjl
* @date 2018年3月9日
*/
public class KylinDialect extends Dialect {
protected static final String SQL_END_DELIMITER = ";";
public String getLimitString(String sql, boolean hasOffset) {
return KylinPageHepler.getLimitString(sql, -1, -1);
}
@Override
public String getLimitString(String sql, int offset, int limit) {
return KylinPageHepler.getLimitString(sql, offset, limit);
}
@Override
public boolean supportsLimit() {
return true;
}
}
/**
* 分頁幫助類
*
* @author zhangjl
* @date 2018年3月9日
*/
public class KylinPageHepler {
/**
* 得到查詢總數的sql
*/
public static String getCountString(String querySelect) {
querySelect = getLineSql(querySelect);
int orderIndex = getLastOrderInsertPoint(querySelect);
int formIndex = getAfterFormInsertPoint(querySelect);
String select = querySelect.substring(0, formIndex);
// 如果SELECT 中包含 DISTINCT 只能在外層包含COUNT
if (select.toLowerCase().indexOf("select distinct") != -1 || querySelect.toLowerCase().indexOf("group by") != -1) {
return new StringBuffer(querySelect.length()).append("select count(1) from (").append(querySelect.substring(0, orderIndex))
.append(" ) t").toString();
} else {
return new StringBuffer(querySelect.length()).append("select count(1) ").append(querySelect.substring(formIndex, orderIndex))
.toString();
}
}
/**
* 得到最后一個Order By的插入點位置
*
* @return 返回最后一個Order By插入點的位置
*/
private static int getLastOrderInsertPoint(String querySelect) {
int orderIndex = querySelect.toLowerCase().lastIndexOf("order by");
if (orderIndex == -1 || !isBracketCanPartnership(querySelect.substring(orderIndex, querySelect.length()))) {
throw new RuntimeException("Kylin SQL 分頁必須要有Order by 語句!");
}
return orderIndex;
}
/**
* 得到分頁的SQL
*
* @param offset 偏移量
* @param limit 位置
* @return 分頁SQL
*/
public static String getLimitString(String querySelect, int offset, int limit) {
querySelect = getLineSql(querySelect);
// String sql = querySelect + " limit " + offset + " ," + limit;
String sql = querySelect + " limit " + limit + " offset " + offset;
return sql;
}
/**
* 將SQL語句變成一條語句，並且每個單詞的間隔都是1個空格
*
* @param sql SQL語句
* @return 如果sql是NULL返回空，否則返回轉化后的SQL
*/
private static String getLineSql(String sql) {
return sql.replaceAll("[\r\n]", " ").replaceAll("\\s{2,}", " ");
}
/**
* 得到SQL第一個正確的FROM的的插入點
*/
private static int getAfterFormInsertPoint(String querySelect) {
String regex = "\\s+FROM\\s+";
Pattern pattern = Pattern.compile(regex, Pattern.CASE_INSENSITIVE);
Matcher matcher = pattern.matcher(querySelect);
while (matcher.find()) {
int fromStartIndex = matcher.start(0);
String text = querySelect.substring(0, fromStartIndex);
if (isBracketCanPartnership(text)) {
return fromStartIndex;
}
}
return 0;
}
/**
* 判斷括號"()"是否匹配,並不會判斷排列順序是否正確
*
* @param text 要判斷的文本
* @return 如果匹配返回TRUE,否則返回FALSE
*/
private static boolean isBracketCanPartnership(String text) {
if (text == null || (getIndexOfCount(text, '(') != getIndexOfCount(text, ')'))) {
return false;
}
return true;
}
/**
* 得到一個字符在另一個字符串中出現的次數
*
* @param text 文本
* @param ch 字符
*/
private static int getIndexOfCount(String text, char ch) {
int count = 0;
for (int i = 0; i < text.length(); i++) {
count = (text.charAt(i) == ch) ? count + 1 : count;
}
return count;
}
}
OK，改造完成，又得一波驗證跟測試。
解決方案二：Freemarker
解決方案驗證沒問題，其實有一個更簡潔的方案。
1 使用 Freemarker 的渲染功能 ，渲染SQL模板（達到類似 Mybatis 的能力）
2 使用數據庫保存 SQL 模板；
3 使用Spring的JdbcTemplate
個人覺得這個方案有比用 Mybatis 更好的優勢，SQL 有問題了，用 DML 改下 SQL 模板，都不用改代碼了！
下面放出一個 Freemaker 渲染 SQL 模板示例代碼：
import java.io.StringWriter;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import freemarker.cache.StringTemplateLoader;
import freemarker.template.Configuration;
import freemarker.template.Template;
/**
* Freemarker 測試
* @author zhangjl
* @date 2018年3月9日
*/
public class FreemarkerTest {
private static final Logger LOG = LoggerFactory.getLogger(FreemarkerTest.class);
@Test
public void test1() {
try {
/**
* <pre>
* SELECT *
FROM shop
WHERE 1 = 1
<#if id??>AND id = ${id?c} </#if>
<#if order??>ORDER BY ${order} <#if sort??> ${sort} </#if> </#if>
* </pre>
*/
String sqlTemplate = "SELECT * FROM shop WHERE 1 = 1 <#if id??>AND id = ${id?c} </#if> <#if order??>ORDER BY ${order} <#if sort??> ${sort} </#if> </#if>";
String sqlId = "shop.list";
Map<String, Object> paramMap = new HashMap<>();
paramMap.put("id", 1L);
String sql = getSql(sqlId, sqlTemplate, paramMap);
LOG.info("解析后的SQL={}", sql);
} catch (Exception e) {
LOG.error("解析SQL異常", e);
}
}
public String getSql(String sqlId, String sqlTemplate, Map<String, Object> paramMap) throws Exception {
StringTemplateLoader loader = new StringTemplateLoader();
loader.putTemplate(sqlId, sqlTemplate);
Configuration conf = new Configuration();
conf.setTemplateLoader(loader);
Template template = conf.getTemplate(sqlId, "utf-8");
StringWriter writer = new StringWriter();
template.process(paramMap, writer);
String sql = writer.toString();
writer.close();
return sql;
}
}