Uncode-DAL是 Java 通用数据访问组件基于mybatis、spring jdbc、hibernate等ORM框架开发

package cn.uncode.dal.core; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.ObjectUtils; import cn.uncode.dal.asyn.AsynContext; import cn.uncode.dal.asyn.AsynSQLTask; import cn.uncode.dal.asyn.Method; import cn.uncode.dal.cache.CacheManager; import cn.uncode.dal.criteria.Model; import cn.uncode.dal.criteria.QueryCriteria; import cn.uncode.dal.criteria.QueryCriteria.Criteria; import cn.uncode.dal.descriptor.Column; import cn.uncode.dal.descriptor.ForeignKey; import cn.uncode.dal.descriptor.QueryResult; import cn.uncode.dal.descriptor.Table; import cn.uncode.dal.descriptor.db.ResolveDataBase; import cn.uncode.dal.descriptor.resolver.JavaType; import cn.uncode.dal.descriptor.resolver.JavaTypeConversion; import cn.uncode.dal.descriptor.resolver.JavaTypeResolver; import cn.uncode.dal.event.EventManager; import cn.uncode.dal.event.asyn.EventContext; import cn.uncode.dal.exception.DalSqlException; import cn.uncode.dal.internal.util.message.Messages; import cn.uncode.dal.router.DefaultMasterSlaveRouter; import cn.uncode.dal.router.MasterSlaveRouter; import cn.uncode.dal.router.TableShardingRouter; import cn.uncode.dal.utils.BeanUtil; import cn.uncode.dal.utils.JsonUtils; import cn.uncode.dal.utils.ShardsUtils; public abstract class AbstractBaseDAL implements BaseDAL { private static final Logger LOG = LoggerFactory.getLogger(AbstractBaseDAL.class); public static final String CACHE_KEY_PREFIX = "uncode_dal_"; private static final String CACHE_KEY_SELECT_BY_CRITERIA = "_selectByCriteria_"; private static final String CACHE_KEY_COUNT_BY_CRITERIA = "_countByCriteria_"; private static final String CACHE_KEY_SELECT_PAGE_BY_CRITERIA = "_selectPageByCriteria_"; private static final String CACHE_KEY_SELECT_BY_IDS = "_selectPageByIds_"; public static final String CACHE_KEY_SELECT_BY_PRIMARY_KEY = "_selectByPrimaryKey_"; protected CacheManager cacheManager; protected ResolveDataBase resolveDatabase; protected MasterSlaveRouter router = new DefaultMasterSlaveRouter(); protected boolean useCache = true; protected String version; protected List<String> versionTables; protected List<String> listenerNames; /** * 采用“需要缓存的”来筛选 */ protected boolean useCacheFilter = true; protected List<String> noCacheTables; protected List<String> cacheTables; //------------------------- // 异步 //------------------------- /** * 日志队列 */ private BlockingQueue<AsynContext> sqlQueue; /** * 写线程池 */ private ExecutorService asynWriterService; /** * 异步日志线程池大小 */ private int asynWriterThreadSize = 5; private List<AsynSQLTask> tasks = new ArrayList<AsynSQLTask>(); public AbstractBaseDAL() { noCacheTables = new ArrayList<String>(); cacheTables = new ArrayList<String>(); sqlQueue = new LinkedBlockingQueue<AsynContext>(); asynWriterService = Executors.newFixedThreadPool(asynWriterThreadSize); // for (int i = 0; i < asynWriterThreadSize; i++) { AsynSQLTask task = new AsynSQLTask(this); task.setLogQueue(sqlQueue); tasks.add(task); asynWriterService.submit(task); // } LOG.info("Asyn dal init ok!"); } //------------------------------------------------------------------- // 抽象方法 //------------------------------------------------------------------- public abstract List<Map<String, Object>> _selectByCriteria(final Table table); public abstract int _countByCriteria(final Table table); public abstract Map<String, Object> _selectByPrimaryKey(final Table table); public abstract long _insert(Table table); public abstract int _insertList(Table table); public abstract int _updateByCriteria(Table table); public abstract int _updateByPrimaryKey(Table table); public abstract int _deleteByPrimaryKey(Table table); public abstract int _deleteByCriteria(Table table); //------------------------------------------------------------------- // 业务核心方法实现-selectPageByCriteria //------------------------------------------------------------------- @Override public QueryResult selectPageByCriteria(QueryCriteria queryCriteria) { return selectPageByCriteria(queryCriteria, 5000); } @Override public QueryResult selectPageByCriteria(String[] fields, QueryCriteria queryCriteria) { return selectPageByCriteria(fields, queryCriteria, 5000); } @Override public QueryResult selectPageByCriteria(List<String> fields, QueryCriteria queryCriteria) { return selectPageByCriteria(fields, queryCriteria, 5000); } @Override public QueryResult selectPageByCriteria(QueryCriteria queryCriteria, int seconds) { String cacheKey = generateMultiSelectCacheKey(null, queryCriteria, CACHE_KEY_SELECT_PAGE_BY_CRITERIA); QueryResult queryResult = null; if (TableShardingRouter.containsTable(queryCriteria.getTable())) { queryResult = ShardsUtils.shardingSelectPageByCriteria(asynWriterService, this, queryCriteria, null); } else { int total = countByCriteria(queryCriteria, seconds); if (total > 0) { int pageCount = total / queryCriteria.getPageSize(); if (total % queryCriteria.getPageSize() != 0) { pageCount++; } if (queryCriteria.getPageIndex() > pageCount) { queryCriteria.setPageIndex(pageCount); } queryResult = selectByCriteria(queryCriteria, seconds); Map<String, Object> page = new HashMap<String, Object>(); page.put(PAGE_INDEX_KEY, queryCriteria.getPageIndex()); page.put(PAGE_SIZE_KEY, queryCriteria.getPageSize()); page.put(PAGE_COUNT_KEY, pageCount); page.put(RECORD_TOTAL_KEY, total); queryResult.setPage(page); } } if (queryResult != null) { if (cacheManager != null && cacheManager.getCache() != null && useCache) { cacheManager.getCache().putObject(buildTableCacheSpace(queryCriteria), cacheKey, queryResult, seconds); } } return queryResult; } @Override public QueryResult selectPageByCriteria(String[] fields, QueryCriteria queryCriteria, int seconds) { QueryResult queryResult = null; String cacheKey = generateMultiSelectCacheKey(null, queryCriteria, CACHE_KEY_SELECT_PAGE_BY_CRITERIA); if (TableShardingRouter.containsTable(queryCriteria.getTable())) { queryResult = ShardsUtils.shardingSelectPageByCriteria(asynWriterService, this, queryCriteria, Arrays.asList(fields)); } else { int total = countByCriteria(queryCriteria, seconds); if (total > 0) { int pageCount = total / queryCriteria.getPageSize(); if (total % queryCriteria.getPageSize() != 0) { pageCount++; } if (queryCriteria.getPageIndex() > pageCount) { queryCriteria.setPageIndex(pageCount); } queryResult = selectByCriteria(fields, queryCriteria, seconds); Map<String, Object> page = new HashMap<String, Object>(); page.put(PAGE_