大数据处理思想

package cn.richinfo.cmail.logger.service.db; import java.sql.Connection; import java.sql.SQLException; import java.util.ArrayList; import java.util.Calendar; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import richinfo.tools.Tools; import cn.richinfo.cmail.logger.common.base.BaseDB; import cn.richinfo.cmail.logger.common.config.CmailLoggerConfigUtils; import cn.richinfo.cmail.logger.common.constant.CommonConstants; import cn.richinfo.cmail.logger.common.exception.dao.DaoException; import cn.richinfo.cmail.logger.common.exception.service.ServiceException; import cn.richinfo.cmail.logger.common.log.CmailLogger; import cn.richinfo.cmail.logger.connManager.ConnManager; import cn.richinfo.cmail.logger.dto.common.LogRecordAddr; import cn.richinfo.cmail.logger.dto.inquiryLog.self.LogData; /** ?* @Description ：跟DB业务相关的基础类，获取分库对应的连接 ?* @date(创建日期)：2012-8-12 下午01:58:53 ?* @company(公司)： ?* ?* @History(修改历史): ?*/ public abstract class BaseDBService extends BaseDB { ?? ? ?? ?private static CmailLogger logger = CmailLogger.getInstance(); ?? ?protected Object doCommand(Long corpId,Object input) throws ServiceException,DaoException { ?? ??? ? ?? ??? ?Connection conn = null; ?? ??? ?try { ?? ??? ??? ?conn = getConn(corpId); ?? ??? ?} catch (SQLException e) { ?? ??? ??? ?logger.error(CmailLogger.CMD_COMMON,CmailLogger.RESULT_FAIL,"获取数据库连接失败",e); ?? ??? ??? ?throw new DaoException("获取数据库连接失败",e); ?? ??? ?} ?? ??? ? ?? ??? ?Object obj = this.doCommand(conn,input); ?? ??? ?closeConn(corpId,conn); ?? ??? ?return obj; ?? ?} ?? ? ?? ?private void closeConn(Long corpId,Connection conn) { ?? ??? ?ConnManager.closeConnection(corpId,conn); ?? ?} ?? ?private Connection getConn(Long corpId) throws SQLException { ?? ??? ?return ConnManager.getConnection(corpId); ?? ?} ?? ? ?? ?private static ConcurrentHashMap<Class<?>,Object> classInstanceMap = new ConcurrentHashMap<Class<?>,Object>(); ?? ? ?? ?/** ?? ? * 获取单例对象 ?? ? * @param className ?? ? * @return ?? ? * @throws ServiceException ?? ? * @throws InstantiationException ?? ? * @throws IllegalAccessException ?? ? */ ?? ?protected static Object getInstance(Class<?> className) throws ServiceException { ?? ??? ? ?? ??? ?Object obj = classInstanceMap.get(className); ?? ??? ? ?? ??? ?if (obj != null) return obj; ?? ??? ? ?? ??? ?synchronized (className) { ?? ??? ??? ?try { ?? ??? ??? ??? ?obj = className.newInstance(); ?? ??? ??? ?} catch (Exception e) { ?? ??? ??? ??? ?throw new ServiceException("类加载失败",e); ?? ??? ??? ?} ?? ??? ??? ?classInstanceMap.put(className,obj); ?? ??? ?} ?? ??? ? ?? ??? ?return obj; ?? ?} ?? ? ?? ? ?? ?/** ?? ? * 具体的业务处理类 ?? ? * @param conn ?? ? * @param input ?? ? * @return ?? ? * @throws Exception ?? ? */ ?? ?protected abstract Object doCommand(Connection conn,DaoException; ?? ?/** ?? ? * 将日志缓存分布解析其对应数据对象 ?? ? * @param logRecordAddr ?? ? * @return ?? ? */ ?? ?protected Map<String,LogRecordAddr> getLogRecordAddrs(String logRecordAddr) { ?? ??? ? ?? ??? ?// key : 日期?? 20120822， value ： 日期与该天的日志记录数 ?? ??? ?Map<String,LogRecordAddr> map = null; ?? ??? ? ?? ??? ?if (Tools.isEmpty(logRecordAddr)) return map; ?? ??? ? ?? ??? ?map = new HashMap<String,LogRecordAddr>(); ?? ??? ? ?? ??? ?String [] s = logRecordAddr.split(CommonConstants.LOG_CACHE_SEPARATOR); ?? ??? ? ?? ??? ?LogRecordAddr recordAddr = null; ?? ??? ?String [] addr = null; ?? ??? ?for (String string : s) { ?? ??? ??? ?addr = string.split(CommonConstants.LOG_CACHE_RECORD_SEPT); ?? ??? ??? ?recordAddr = new LogRecordAddr(); ?? ??? ??? ?recordAddr.setRecordAddr(addr[0]); ?? ??? ??? ?recordAddr.setCount(Integer.valueOf(addr[1])); ?? ??? ??? ? ?? ??? ??? ?map.put(recordAddr.getRecordAddr(),recordAddr); ?? ??? ?} ?? ??? ? ?? ??? ?return map; ?? ?} ?? ? ?? ?/** ?? ? * 将日志缓存分布解析其对应数据对象 ?? ? * @return ?? ? */ ?? ?protected Map<String,LogRecordAddr> getLogRecordAddrs(String agentMailAddrs,String agentMailDetailAddr,String inputAgentAddr) { ?? ? ?? ??? ?// key : 日期?? 20120822， value ： 日期与该天的日志记录数 ?? ??? ?Map<String,LogRecordAddr> map = null; ?? ??? ? ?? ??? ?if (Tools.isEmpty(agentMailDetailAddr) || Tools.isEmpty(agentMailAddrs)) return map; ?? ??? ? ?? ??? ?String [] addrs = agentMailAddrs.split(CommonConstants.LOG_CACHE_SEPARATOR); ?? ??? ?String key = ""; ?? ??? ?String value = ""; ?? ??? ?for (int i = 0; i < addrs.length; i++) { ?? ??? ??? ?value = addrs[i].substring(0,addrs[i].indexOf(CommonConstants.LOG_CACHE_RECORD_SEPT)); ?? ??? ??? ?key = addrs[i].substring(value.length() + 1,addrs[i].length()); ?? ??? ??? ?if (key.equals(inputAgentAddr)) { ?? ??? ??? ??? ?break; ?? ??? ??? ?} ?? ??? ?} ?? ??? ?if (Tools.isEmpty(key)) return map; ?? ??? ? ?? ??? ?map = new HashMap<String,LogRecordAddr>(); ?? ??? ? ?? ??? ?String [] s = agentMailDetailAddr.split(CommonConstants.LOG_CACHE_SEPARATOR); ?? ??? ? ?? ??? ?LogRecordAddr recordAddr = null; ?? ??? ?String [] addr = null; ?? ??? ?for (String string : s) { ?? ??? ??? ?addr = string.split(CommonConstants.LOG_CACHE_RECORD_SEPT); ?? ??? ??? ?if (addr[0].equals(value)) {// 跟某个代收邮件地址对应的序列号相匹配 ?? ??? ??? ??? ?recordAddr = new LogRecordAddr(); ?? ??? ??? ??? ?recordAddr.setRecordAddr(addr[1]); ?? ??? ??? ??? ?recordAddr.setCount(Integer.valueOf(addr[2])); ?? ??? ??? ??? ?map.put(recordAddr.getRecordAddr(),recordAddr); ?? ??? ??? ?} ?? ??? ?} ?? ??? ? ?? ??? ?return map; ?? ?} ?? ?/** ?? ? * 根据将日志缓存分布解析成的数据对象统计出历史日志总和 ?? ? * @param list ?? ? * @return ?? ? */ ?? ?protected Long countFromRecoredAddr(Map<String,LogRecordAddr> map) { ?? ??? ? ?? ??? ?Long totalCount = new Long(0); ?? ??? ?if (map == null || map.size() == 0) return totalCount; ?? ??? ? ?? ??? ?for (String key : map.keySet()) { ?? ??? ??? ?totalCount += map.get(key).getCount(); ?? ??? ?} ?? ??? ? ?? ??? ?return totalCount; ?? ?} ?? ?/** ?? ? * 获取要查询的地址分布：具体从哪几张表中查询数据，查询表中的第几条记录（定位开始索引和结束索引） ?? ? * @param logData 属性中有分页信息 ?? ? * @param recoredAddrMap 日志分布MAP ?? ? * @param countToday 当天日志数 ?? ? * @return ?? ? */ ?? ?public List<LogRecordAddr> listRecordAddr(LogData logData,Map<String,LogRecordAddr> recoredAddrMap,Long countToday) { ?? ??? ?/* ?? ??? ? * 1，先把落到分页范围之内的记录拿出来，放入list中 ?? ??? ? * 2，对list中的每张表设置查询起始和查询终止索引 ?? ??? ? */ ?? ??? ?List<LogRecordAddr> result = new ArrayList<LogRecordAddr>(); ?? ??? ? ?? ??? ?// 从第几条记录开始 ?? ??? ?Long startRecoredIndex = (logData.getCurrentPage() - 1) * logData.getPageSize() + 1; ?? ??? ?Long endRecoredIndex = startRecoredIndex + logData.getPageSize() - 1; ?? ??? ? ?? ??? ?Long flagCount = new Long(0);// 统计到数据之前的记录数(不含当次) ?? ??? ?Long count = new Long(0);// 统计到第几条的记录数 ?? ??? ?for (int i = 0; i <= CmailLoggerConfigUtils.LOG_CONFIG.getLogDB().getSaveLogDay(); i++) { ?? ??? ??? ? ?? ??? ??? ?LogRecordAddr logRecordAddr = null; ?? ??? ??? ? ?? ??? ??? ?Long currentCount = new Long(0); ?? ??? ??? ?if (i == 0) { ?? ??? ??? ??? ?currentCount = countToday; ?? ??? ??? ??? ?logRecordAddr = new LogRecordAddr(); ?? ??? ??? ??? ?logRecordAddr.setCount(countToday.intValue()); ?? ??? ??? ??? ?logRecordAddr.setRecordAddr(this.formatDateTime(Calendar.getInstance().getTime())); ?? ??? ??? ?} else { ?? ??? ??? ??? ? ?? ??? ??? ??? ?if (recoredAddrMap == null) break; ?? ??? ??? ??? ? ?? ??? ??? ??? ?Calendar c = Calendar.getInstance(); ?? ??? ??? ??? ?c.add(Calendar.DAY_OF_YEAR,-i); ?? ??? ??? ??? ?logRecordAddr = recoredAddrMap.get(this.formatDateTime(c.getTime())); ?? ??? ??? ??? ?if (logRecordAddr == null) continue;// 倒数第i天没有数据，返回 ?? ??? ??? ??? ?currentCount = new Long(logRecordAddr.getCount()); ?? ??? ??? ?} ?? ??? ??? ? ?? ??? ??? ?count += currentCount; ?? ??? ??? ? ?? ??? ??? ?if (startRecoredIndex > count) { ?? ??? ??? ??? ?flagCount += currentCount; ?? ??? ??? ??? ?continue; ?? ??? ??? ?} ?? ??? ??? ? ?? ??? ??? ?if (startRecoredIndex <= count && endRecoredIndex >= count) {// 本次 + 取next ?? ??? ??? ??? ?result.add(logRecordAddr); ?? ??? ??? ??? ?continue; ?? ??? ??? ?} ?? ??? ??? ? ?? ??? ??? ?if (count >= endRecoredIndex) {// 终止的记录 ?? ??? ??? ??? ?result.add(logRecordAddr); ?? ??? ??? ??? ?break; ?? ??? ??? ?} ?? ??? ??? ? ?? ??? ?} ?? ??? ? ?? ????? ?? ??? ?if (fetchTab.size() == 1) { ?? ??? ??? ?fetchTab.get(0).setStartIndex(startRecoredIndex - flagCount); ?? ??? ??? ?fetchTab.get(0).setEndIndex(fetchTab.get(0).getStartIndex() + pageSize - 1); ?? ??? ?} else if (fetchTab.size() == 2) { ?? ??? ??? ?fetchTab.get(0).setStartIndex(startRecoredIndex - flagCount); ?? ??? ??? ?fetchTab.get(0).setEndIndex(new Long(fetchTab.get(0).getCount())); ?? ??? ??? ?fetchTab.get(1).setStartIndex(new Long(0)); ?? ??? ??? ?fetchTab.get(1).setEndIndex(pageSize - (fetchTab.get(0).getEndIndex() - result.get(0).getStartIndex() + 1)); ?? ??? ??? ?//update ?? ??? ??? ?fetchTab.get(1).setPageSize(fetchTab.get(1).getEndIndex()); ?? ??? ??? ? ?? ??? ?} else if (fetchTab.size() > 2){ ?? ??? ??? ?int resultCount = 0; ?? ??? ??? ?for (int i = 0; i < fetchTab.size(); i++) { ?? ??? ??? ??? ?if (i == 0) { ?? ??? ??? ??? ??? ?fetchTab.get(0).setStartIndex(startRecoredIndex - flagCount); ?? ??? ??? ??? ??? ?fetchTab.get(0).setEndIndex(new Long(fetchTab.get(0).getCount())); ?? ??? ??? ??? ?} else if (i == fetchTab.size() -1 ) { ?? ??? ??? ??? ??? ?fetchTab.get(i).setStartIndex(new Long(0)); ?? ??? ??? ??? ??? ?fetchTab.get(i).setEndIndex(new Long(pageSize - resultCount)); ?? ??? ??? ??? ??? ?fetchTab.get(i).setPageSize(fetchTab.get(i).getEndIndex()); ?? ??? ??? ??? ?} else { ?? ??? ??? ??? ??? ?fetchTab.get(i).setStartIndex(new Long(0)); ?? ??? ??? ??? ??? ?fetchTab.get(i).setEndIndex(new Long(fetchTab.get(i).getCount())); ?? ??? ??? ??? ?} ?? ??? ??? ??? ?long endIndex = fetchTab.get(i).getEndIndex(); ?? ??? ??? ??? ?long startIndex = fetchTab.get(i).getStartIndex(); ?? ??? ??? ??? ?resultCount += endIndex - startIndex + 1; ?? ??? ??? ?} ?? ??? ?} ?? ??? ? ?? ??? ?return result; ?? ?} ?? ?protected Long getPageCount(Long totalCount,Integer pageSize) { ?? ??? ? ?? ??? ?long pageCount = 0; ?? ??? ? ?? ??? ?if (totalCount == 0) return pageCount; ?? ??? ? ?? ??? ?if(totalCount % pageSize == 0){ ?? ??? ??? ?pageCount = totalCount / pageSize; ?? ??? ?} else { ?? ??? ??? ?pageCount = totalCount / pageSize + 1; ?? ??? ?}?? ? ?? ??? ? ?? ??? ?return pageCount; ?? ?} }