diff --git a/src/main/java/com/nis/util/Constants.java b/src/main/java/com/nis/util/Constants.java index 7489828..e2a4ed7 100644 --- a/src/main/java/com/nis/util/Constants.java +++ b/src/main/java/com/nis/util/Constants.java @@ -219,9 +219,13 @@ public final class Constants { /** - * redis分布式锁超时时间,默认五分钟,3000秒 + * redis分布式锁超时时间,默认五分钟,300秒 */ - public static final Long REDISLOCKTIME=Configurations.getLongProperty("redisLockTime", 3000); + public static final Long REDISLOCKTIME=Configurations.getLongProperty("redisLockTime", 300); + /** + * 全量数据同步分布式锁时间,默认两个小时 + */ + public static final Long CONFIGSYNCLOCKTIME=Configurations.getLongProperty("configSyncLockTime", 7200); /** * 获取redis分布式锁失败后的尝试获取锁的次数,每次失败暂停一秒钟后再次尝试 */ diff --git a/src/main/java/com/nis/util/JedisClusterUtils.java b/src/main/java/com/nis/util/JedisClusterUtils.java new file mode 100644 index 0000000..e5d1044 --- /dev/null +++ b/src/main/java/com/nis/util/JedisClusterUtils.java @@ -0,0 +1,361 @@ +package com.nis.util; + +import java.util.Collections; +import java.util.List; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.Lists; +import com.nis.restful.RestBusinessCode; +import com.nis.restful.ServiceRuntimeException; +import com.nis.web.service.SpringContextHolder; + +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.exceptions.JedisException; + +/** + * @ClassName: JedisClusterUtils + * @Description: redis-cluster库工具类 + * @author (rkg) + * @date 2018年12月01日 上午10:31:00 + * @version V1.0 + */ +public class JedisClusterUtils { + private static Logger logger = LoggerFactory.getLogger(JedisClusterUtils.class); + + /** + * 获取缓存 + * + * @param key 键 + * @return 值 + */ + public static String get(String key) { + String value = null; + JedisCluster jedis = null; + try { + jedis = getResource(); + if (jedis.exists(key)) { + value = jedis.get(key); + value = StringUtils.isNotBlank(value) && !"nil".equalsIgnoreCase(value) ? value : null; + logger.debug("get {} = {}", key, value); + } + } catch (Exception e) { + throw new ServiceRuntimeException("从redis-cluster中获取" + key + "对应的值失败", + RestBusinessCode.KeyNotExistsInRedis.getValue()); + } finally { + returnResource(jedis); + } + return value; + } + + /** + * 获取缓存 + * + * @param key 键 + * @return 值 + */ + public static Object getObject(String key) { + Object value = null; + JedisCluster jedis = null; + try { + jedis = getResource(); + if (jedis.exists(getBytesKey(key))) { + value = toObject(jedis.get(getBytesKey(key))); + logger.debug("getObject {} = {}", key, value); + } + } catch (Exception e) { + throw new ServiceRuntimeException("从redis-cluster中获取" + key + "对应的值失败", + RestBusinessCode.KeyNotExistsInRedis.getValue()); + } finally { + returnResource(jedis); + } + return value; + } + + /** + * 可以作为获取唯一id的方法
+ * 将key对应的value加上指定的值，只有value可以转为数字时该方法才可用 + * + * @param String key + * @param long number 要减去的值 + * @return long 相加后的值 + */ + public static long incrBy(String key, long number) { + JedisCluster jedis = getResource(); + long len = jedis.incrBy(key, number); + returnResource(jedis); + return len; + } + + /** + * 设置缓存 + * + * @param key 键 + * @param value 值 + * @param cacheSeconds 超时时间，0为不超时 + * @return + */ + public static String set(String key, String value, int cacheSeconds) { + String result = null; + JedisCluster jedis = null; + try { + jedis = getResource(); + result = jedis.set(key, value); + if (cacheSeconds != 0) { + jedis.expire(key, cacheSeconds); + } + logger.debug("set {} = {}", key, value); + } catch (Exception e) { + throw new ServiceRuntimeException("向redis-cluster中设置zset失败,key=" + key + ",value=" + value, + RestBusinessCode.ZsetFailed.getValue()); + } finally { + returnResource(jedis); + } + return result; + } + + /** + * 获取List缓存 + * + * @param key 键 + * @return 值 + */ + public static List getList(String key) { + List value = null; + JedisCluster jedis = null; + try { + jedis = getResource(); + if (jedis.exists(key)) { + value = jedis.lrange(key, 0, -1); + logger.debug("getList {} = {}", key, value); + } + } catch (Exception e) { + throw new ServiceRuntimeException("从redis-cluster中获取" + key + "对应的值失败", + RestBusinessCode.KeyNotExistsInRedis.getValue()); + } finally { + returnResource(jedis); + } + return value; + } + + /** + * 获取List缓存 + * + * @param key 键 + * @return 值 + */ + public static List getObjectList(String key) { + List value = null; + JedisCluster jedis = null; + try { + jedis = getResource(); + if (jedis.exists(getBytesKey(key))) { + List list = jedis.lrange(getBytesKey(key), 0, -1); + value = Lists.newArrayList(); + for (byte[] bs : list) { + value.add(toObject(bs)); + } + logger.debug("getObjectList {} = {}", key, value); + } + } catch (Exception e) { + throw new ServiceRuntimeException("从redis-cluster中获取" + key + "对应的值失败", + RestBusinessCode.KeyNotExistsInRedis.getValue()); + } finally { + returnResource(jedis); + } + return value; + } + + /** + * 缓存是否存在 + * + * @param key 键 + * @return + */ + public static boolean exists(String key) { + boolean result = false; + JedisCluster jedis = null; + try { + jedis = getResource(); + result = jedis.exists(key); + logger.debug("exists {}", key); + } catch (Exception e) { + throw new ServiceRuntimeException("从redis-cluster中判断" + key + "是否存在失败", + RestBusinessCode.ExistsKeyFailed.getValue()); + } finally { + returnResource(jedis); + } + return result; + } + + /** + * 缓存是否存在 + * + * @param key 键 + * @return + */ + public static boolean existsObject(String key) { + boolean result = false; + JedisCluster jedis = null; + try { + jedis = getResource(); + result = jedis.exists(getBytesKey(key)); + logger.debug("existsObject {}", key); + } catch (Exception e) { + throw new ServiceRuntimeException("从redis-cluster中判断" + key + "是否存在失败", + RestBusinessCode.ExistsKeyFailed.getValue()); + } finally { + returnResource(jedis); + } + return result; + } + + /** + * 获取资源 + * + * @return + * @throws JedisException + */ + public static JedisCluster getResource() throws JedisException { + JedisCluster jedisCluster = SpringContextHolder.getBean(JedisCluster.class); + if (jedisCluster == null) { + throw new ServiceRuntimeException("无法获取redis-cluster连接,请联系管理员检查程序", + RestBusinessCode.CannotConnectionRedis.getValue()); + } + return jedisCluster; + + } + + /** + * 释放资源 + * + * @param jedis + * @param isBroken + */ + public static void returnBrokenResource(JedisCluster jedis) { + if (jedis != null) { + try { + jedis.close(); + } catch (Exception e) { + throw new ServiceRuntimeException("释放redis-cluster连接失败", RestBusinessCode.CannotConnectionRedis.getValue()); + } + } + } + + /** + * 释放资源 + * + * @param jedis + * @param isBroken + */ + public static void returnResource(JedisCluster jedis) { + if (jedis != null) { + try { + jedis.close(); + } catch (Exception e) { + throw new ServiceRuntimeException("释放redis-cluster连接失败", RestBusinessCode.CannotConnectionRedis.getValue()); + } + } + } + + /** + * 获取byte[]类型Key + * + * @param key + * @return + */ + public static byte[] getBytesKey(Object object) { + if (object instanceof String) { + return StringUtils.getBytes((String) object); + } else { + return ObjectUtils.serialize(object); + } + } + + /** + * Object转换byte[]类型 + * + * @param key + * @return + */ + public static byte[] toBytes(Object object) { + return ObjectUtils.serialize(object); + } + + /** + * byte[]型转换Object + * + * @param key + * @return + */ + public static Object toObject(byte[] bytes) { + return ObjectUtils.unserialize(bytes); + } + + // 设置成功返回的结果OK + private static final String LOCK_SUCCESS = "OK"; + // NX -- Only set the key if it does not already exist. XX -- Only set the key + // if it already exist + private static final String SET_IF_NOT_EXIST = "NX"; + // 失效单位秒(EX)还是毫秒(PX) + private static final String SET_WITH_EXPIRE_TIME = "EX"; + private static final Long UNLOCK_SUCCESS = 1L; + + /** + * 尝试获取分布式锁,如果没有key就set,有key就不操作 + * + * @param requestId 请求标识(UUID.randomUUID().toString()),正常情况下是谁加的锁,谁去解锁不能a加的锁,b去解锁 + * @return 是否获取成功 + */ + public static Boolean lock(String requestId) { + String key = "uiAndServiceConfigSyncLock"; + String var1 = getResource().set(key, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, + Constants.REDISLOCKTIME); + if (LOCK_SUCCESS.equals(var1)) { + return true; + } + return false; + } + + /** + * 解锁操作 + * + * @param value 客户端标识(requestId) + * @return + */ + public static Boolean unLock(String value) { + String key = "uiAndServiceConfigSyncLock"; + // 这个字符串是个lua脚本，代表的意思是如果根据key拿到的value跟传入的value相同就执行del，否则就返回0【保证安全性】 + String luaScript = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then return redis.call(\"del\",KEYS[1]) else return 0 end"; + // 这个命令就是去执行lua脚本，KEYS的集合就是第二个参数，ARGV的集合就是第三参数【保证解锁的原子操作】 + Object var2 = getResource().eval(luaScript, Collections.singletonList(key), Collections.singletonList(value)); + if (UNLOCK_SUCCESS == var2) { + return true; + } + return false; + } + + /** + * 重试机制 + * + * @param value 客户端标识 + * @return + */ + public static Boolean lockRetry(String value) { + Boolean flag = false; + try { + for (int i = 0; i < Constants.REDISRETRYNUM; i++) { + flag = lock(value); + if (flag) { + break; + } + Thread.sleep(1000); + } + } catch (Exception e) { + logger.error("尝试获取redis分布式锁失败,失败原因:{}", ExceptionUtil.getExceptionMsg(e)); + } + return flag; + } + +} diff --git a/src/main/java/com/nis/util/JedisUtils.java b/src/main/java/com/nis/util/JedisUtils.java index 54ea353..1fc18b5 100644 --- a/src/main/java/com/nis/util/JedisUtils.java +++ b/src/main/java/com/nis/util/JedisUtils.java @@ -14,7 +14,13 @@ import com.google.common.collect.Lists; import com.nis.restful.RestBusinessCode; import com.nis.restful.ServiceRuntimeException; import com.nis.web.service.SpringContextHolder; - +/** + * @ClassName: JedisUtils + * @Description: redis工具类 + * @author (rkg) + * @date 2018年03月5日 上午10:20:33 + * @version V1.0 + */ public class JedisUtils { private static Logger logger = LoggerFactory.getLogger(JedisUtils.class); private static final JedisPool jedisPool = SpringContextHolder.getBean(JedisPool.class); diff --git a/src/main/java/com/nis/web/controller/restful/ConfigSourcesController.java b/src/main/java/com/nis/web/controller/restful/ConfigSourcesController.java index 7c43803..140cd23 100644 --- a/src/main/java/com/nis/web/controller/restful/ConfigSourcesController.java +++ b/src/main/java/com/nis/web/controller/restful/ConfigSourcesController.java @@ -11,8 +11,6 @@ import java.util.UUID; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; -import net.sf.json.JSONObject; - import org.apache.commons.codec.digest.DigestUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestBody; @@ -38,13 +36,14 @@ import com.nis.web.service.AuditLogThread; import com.nis.web.service.ServicesRequestLogService; import com.nis.web.service.fdfs.FastDFSFile; import com.nis.web.service.fdfs.FileManager; -import com.nis.web.service.restful.ConfigRedisService; import com.nis.web.service.restful.ConfigSourcesService; import com.wordnik.swagger.annotations.Api; import com.wordnik.swagger.annotations.ApiOperation; import com.wordnik.swagger.annotations.ApiParam; import com.zdjizhi.utils.StringUtil; +import net.sf.json.JSONObject; + /** * @ClassName: ConfigSourcesController * @Description: 配置存储服务 @@ -63,9 +62,6 @@ public class ConfigSourcesController extends BaseRestController { @Autowired protected ServicesRequestLogService servicesRequestLogService; - @Autowired - ConfigRedisService configRedisService; - @RequestMapping(value = "/cfg/v1/configSources", method = RequestMethod.POST) @ApiOperation(value = "MAAT规则存储接口", httpMethod = "POST", response = Map.class, notes = "接收MAAT规则数据，存储到流量处理平台配置线中") @ApiParam(value = "MAAT规则对象", name = "configSource", required = true) @@ -549,69 +545,81 @@ public class ConfigSourcesController extends BaseRestController { @RequestMapping(value = "/cfg_batch/v1/configSources", method = RequestMethod.POST, produces = org.springframework.http.MediaType.APPLICATION_JSON_VALUE) @ApiOperation(value = "全量同步配置接收接口", httpMethod = "POST", response = Map.class, notes = "接收全量同步配置") public Map receiveConfigSources(@RequestBody String jsonString, HttpServletRequest request, - HttpServletResponse response) { + HttpServletResponse response) { long start = System.currentTimeMillis(); - AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_POST, request, - null); + AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_POST, request, null); thread.setContent("全量同步不记录请求内容"); try { - + String configType = request.getHeader("Config-Type"); String serviceId = request.getHeader("Service-Id"); String configTable = request.getHeader("Config-Table"); String lastCompletedTag = request.getHeader("Last-Completed-Tag"); - logger.info(new Date()+"-----------接收到json格式数据:"+jsonString+"-------:"); - if(StringUtil.isEmpty(serviceId)){ + logger.info(new Date() + "-----------接收到json格式数据:" + jsonString + "-------:"); + if (StringUtil.isEmpty(serviceId)) { logger.error("未在请求头中获取到serviceId"); - throw new RestServiceException(RestBusinessCode.config_integrity_error.getErrorReason() + "," + "未在请求头中获取到serviceId", - RestBusinessCode.config_integrity_error.getValue()); + throw new RestServiceException( + RestBusinessCode.config_integrity_error.getErrorReason() + "," + "未在请求头中获取到serviceId", + RestBusinessCode.config_integrity_error.getValue()); } + if (StringUtil.isEmpty(configType)) { + logger.error("未在请求头中获取到Config-Type"); + throw new RestServiceException( + RestBusinessCode.config_integrity_error.getErrorReason() + "," + "未在请求头中获取到Config-Type", + RestBusinessCode.config_integrity_error.getValue()); + } else { + if (!("1".equals(configType) || "2".equals(configType))) { + logger.error("Config-Type的值只能是1(maat)和2(回调)"); + throw new RestServiceException( + RestBusinessCode.config_integrity_error.getErrorReason() + "," + + "Config-Type的值只能是1(maat)和2(回调)", + RestBusinessCode.config_integrity_error.getValue()); + + } + } + logger.info("-----------开始存储到json格式数据------->>configType:" + configType + ",serviceId:" + serviceId + + ",configTable:" + configTable + ",lastCompletedTag:" + lastCompletedTag); + String key = null; + if ("1".equals(configType)) { + key = "MAAT"; + } else { + key = "UNMAAT"; + } + key = key + "-" + serviceId + "-" + UUID.randomUUID(); + configSourcesService.setRedisClusterKey(key, jsonString); + configSourcesService.setAllServiceKey(key); if (!StringUtil.isEmpty(lastCompletedTag)) { - System.out.println("接收全量同步配置:FINISHED"); + // 设置配置同步状态为接收配置完成 + configSourcesService.setAllConfigSyncStatus("1"); logger.info("接收全量同步配置:FINISHED"); } - if(StringUtil.isEmpty(configType)){ - logger.error("未在请求头中获取到Config-Type"); - throw new RestServiceException(RestBusinessCode.config_integrity_error.getErrorReason() + "," + "未在请求头中获取到Config-Type", - RestBusinessCode.config_integrity_error.getValue()); - }else{ - if (!("1".equals(configType)||"2".equals(configType))) { - logger.error("Config-Type的值只能是1(maat)和2(回调)"); - throw new RestServiceException(RestBusinessCode.config_integrity_error.getErrorReason() + "," + "Config-Type的值只能是1(maat)和2(回调)", - RestBusinessCode.config_integrity_error.getValue()); - - } - } - - logger.info("-----------开始存储到json格式数据------->>configType:"+configType+",serviceId:"+serviceId+",configTable:"+configTable+",lastCompletedTag:"+lastCompletedTag); -// } catch (Exception e) { // TODO: handle exception logger.error(e.getMessage()); - throw new RestServiceException(RestBusinessCode.config_integrity_error.getErrorReason() + "," + e.getMessage(), + throw new RestServiceException( + RestBusinessCode.config_integrity_error.getErrorReason() + "," + e.getMessage(), RestBusinessCode.config_integrity_error.getValue()); } return compileServiceResponse(thread, System.currentTimeMillis() - start, request, response, "已接收全量同步配置信息", Constants.IS_DEBUG ? jsonString : null); } + @RequestMapping(value = "/cfg_batch/v1/status", method = RequestMethod.POST, produces = org.springframework.http.MediaType.APPLICATION_JSON_VALUE) @ApiOperation(value = "配置全量同步指令下发接口", httpMethod = "POST", response = Map.class, notes = "下发配置同步指令") - public Map acceptStatus(@RequestBody String jsonString, HttpServletRequest request, - HttpServletResponse response) { + public Map acceptStatus(@RequestBody String jsonString, HttpServletRequest request, HttpServletResponse response) { long start = System.currentTimeMillis(); - AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_POST, request, - null); - JSONObject obj= JSONObject.fromObject(jsonString); - - if (!StringUtil.isEmpty(obj.get("syncStatus"))&&"1".equals(obj.get("syncStatus").toString())) { - logger.info("-----------配置同步指令下发:"+new Date()); - }else{ + AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_POST, request, null); + JSONObject obj = JSONObject.fromObject(jsonString); + if (!StringUtil.isEmpty(obj.get("syncStatus")) && "1".equals(obj.get("syncStatus").toString())) { + logger.info("-----------配置同步指令下发:" + new Date()); + // 设置配置同步状态为开始 + configSourcesService.setAllConfigSyncStatus("0"); + } else { logger.error("未获取到同步状态码"); thread.setBusinessCode(RestBusinessCode.syncStatusFailed.getValue()); throw new RestServiceException("未获取到同步状态码", RestBusinessCode.syncStatusFailed.getValue()); } - thread.setBusinessCode(RestBusinessCode.syncStatusSuccessed.getValue()); return compileServiceResponse(thread, System.currentTimeMillis() - start, request, response, "配置同步指令下发成功", Constants.IS_DEBUG ? jsonString : null); @@ -619,21 +627,19 @@ public class ConfigSourcesController extends BaseRestController { @RequestMapping(value = "/cfg_batch/v1/status", method = RequestMethod.GET) @ApiOperation(value = "获取全量同步状态服务接口", httpMethod = "GET", response = Map.class, notes = "获取全量同步状态服务") - public Map getSyncStatus(HttpServletRequest request, - HttpServletResponse response) { + public Map getSyncStatus(HttpServletRequest request, HttpServletResponse response) { long start = System.currentTimeMillis(); - AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_GET, request, - null); + AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_GET, request, null); List list = new ArrayList(); JSONObject obj = new JSONObject(); obj.put("service", "ntc"); - obj.put("status", 0); + obj.put("status", configSourcesService.getAllConfigSyncStatus()); obj.put("opTime", "2018-11-23 08:31:27"); list.add(obj); return compileServiceResponse(thread, System.currentTimeMillis() - start, request, response, "获取全量同步状态成功", list); } - + private void validateGroupReuseSource(AuditLogThread thread, long start, GroupReuseSource groupReuseSource) { String errorInfo = ""; @@ -668,4 +674,15 @@ public class ConfigSourcesController extends BaseRestController { return false; } + @RequestMapping(value = "/cfg/v1/getConfigCount", method = RequestMethod.GET) + @ApiOperation(value = "获取有效无效的配置个数", httpMethod = "GET", response = Map.class, notes = "获取有效无效的配置个数") + public Map getConfigCount(HttpServletRequest request, HttpServletResponse response) { + long start = System.currentTimeMillis(); + AuditLogThread thread = super.saveRequestLog(servicesRequestLogService, Constants.OPACTION_GET, request, null); + Map allConfigByScan = configSourcesService.getAllConfigByScan(); + allConfigByScan.putAll(configSourcesService.getAllConfig()); + return compileServiceResponse(thread, System.currentTimeMillis() - start, request, response, "获取有效无效的配置个数成功", +// configSourcesService.getAllConfig()); + allConfigByScan); + } } diff --git a/src/main/java/com/nis/web/service/restful/ConfigSourcesService.java b/src/main/java/com/nis/web/service/restful/ConfigSourcesService.java index 1758443..453e386 100644 --- a/src/main/java/com/nis/web/service/restful/ConfigSourcesService.java +++ b/src/main/java/com/nis/web/service/restful/ConfigSourcesService.java @@ -8,16 +8,22 @@ import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Matcher; import java.util.regex.Pattern; import net.sf.json.JSONObject; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisCluster; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.ScanParams; +import redis.clients.jedis.ScanResult; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -42,6 +48,8 @@ import com.nis.util.CamelUnderlineUtil; import com.nis.util.CompileVal; import com.nis.util.Constants; import com.nis.util.GroupReuseVal; +import com.nis.util.JedisClusterUtils; +import com.nis.util.JedisUtils; import com.nis.util.JsonMapper; import com.nis.util.ReadCommSourceXmlUtil; import com.nis.util.ServiceAndRDBIndexReal; @@ -126,8 +134,8 @@ public class ConfigSourcesService extends BaseService { StringBuffer sb) throws Exception { Map> maatMap = new HashMap>(); Map> configMap = new HashMap>(); - - if (configCompileList!=null&&configCompileList.size()>Constants.MAX_LIST_SIZE) { + + if (configCompileList != null && configCompileList.size() > Constants.MAX_LIST_SIZE) { thread.setSaveContentFlag(false); } for (ConfigCompile configCompile : configCompileList) { @@ -326,7 +334,7 @@ public class ConfigSourcesService extends BaseService { public void updateConfigSources(AuditLogThread thread, long start, List compileList, Date opTime, StringBuffer sb) throws Exception { Map> compileMap = new HashMap>(); - if (compileList!=null&&compileList.size()>Constants.MAX_LIST_SIZE) { + if (compileList != null && compileList.size() > Constants.MAX_LIST_SIZE) { thread.setSaveContentFlag(false); } if (null != compileList && compileList.size() > 0) { @@ -422,7 +430,7 @@ public class ConfigSourcesService extends BaseService { throw new RestServiceException(RestBusinessCode.CBParamFormateError.getErrorReason() + "," + e.getMessage(), RestBusinessCode.CBParamFormateError.getValue()); } - if (jsonObjectList!=null&&jsonObjectList.size()>Constants.MAX_LIST_SIZE) { + if (jsonObjectList != null && jsonObjectList.size() > Constants.MAX_LIST_SIZE) { thread.setSaveContentFlag(false); } Map>> dstMaps = new HashMap>>(); @@ -495,7 +503,7 @@ public class ConfigSourcesService extends BaseService { maatTableName.substring(maatTableName.lastIndexOf("_") + 1)); dstStr = dstPath.replace("{fileName}", dstStr.substring(dstStr.lastIndexOf("/") + 1)); } else if ("file_id".equals(commonSourceFieldCfg.getDstName())) { - //dstStr = dstStr.substring(dstStr.indexOf("group")); + // dstStr = dstStr.substring(dstStr.indexOf("group")); } } switch (commonSourceFieldCfg.getFieldType()) { @@ -644,7 +652,7 @@ public class ConfigSourcesService extends BaseService { throw new RestServiceException(RestBusinessCode.CBParamFormateError.getErrorReason() + "," + e.getMessage(), RestBusinessCode.CBParamFormateError.getValue()); } - if (jsonObjectList!=null&&jsonObjectList.size()>Constants.MAX_LIST_SIZE) { + if (jsonObjectList != null && jsonObjectList.size() > Constants.MAX_LIST_SIZE) { thread.setSaveContentFlag(false); } // @@ -878,18 +886,21 @@ public class ConfigSourcesService extends BaseService { } else { maatConfig.getIpRegionMapList().addAll(dstMapList); } - - if ((maatConfig.getStrRegionMapList()!=null&&maatConfig.getStrRegionMapList().size()>Constants.MAX_LIST_SIZE) - ||(maatConfig.getStrStrRegionMapList()!=null&&maatConfig.getStrStrRegionMapList().size()>Constants.MAX_LIST_SIZE) - ||(maatConfig.getIpRegionMapList()!=null&&maatConfig.getIpRegionMapList().size()>Constants.MAX_LIST_SIZE) - ||(maatConfig.getNumRegionMapList()!=null&&maatConfig.getNumRegionMapList().size()>Constants.MAX_LIST_SIZE)) { + + if ((maatConfig.getStrRegionMapList() != null + && maatConfig.getStrRegionMapList().size() > Constants.MAX_LIST_SIZE) + || (maatConfig.getStrStrRegionMapList() != null + && maatConfig.getStrStrRegionMapList().size() > Constants.MAX_LIST_SIZE) + || (maatConfig.getIpRegionMapList() != null + && maatConfig.getIpRegionMapList().size() > Constants.MAX_LIST_SIZE) + || (maatConfig.getNumRegionMapList() != null + && maatConfig.getNumRegionMapList().size() > Constants.MAX_LIST_SIZE)) { thread.setSaveContentFlag(false); } - //maatConfig.setService(groupReuse.getService()); + // maatConfig.setService(groupReuse.getService()); list.add(maatConfig); } - - + // 调用接口入redis logger.info("---------------调用Redis 分组复用配置新增接口---------------------"); configRedisService.saveGroupReuseConfig(list); @@ -957,9 +968,9 @@ public class ConfigSourcesService extends BaseService { } } logger.info("调用接口删除Redis中分组复用的域配置接口"); - if (reuseMap!=null&&reuseMap.size()>Constants.MAX_LIST_SIZE) { + if (reuseMap != null && reuseMap.size() > Constants.MAX_LIST_SIZE) { thread.setSaveContentFlag(false); - } + } // 所有的都删除成功返回true if (!configRedisService.delGroupReuseConfig(reuseMap)) { // if (!configRedisService.delGroupReuseConfigByPipeline(reuseMap)) { @@ -1040,4 +1051,90 @@ public class ConfigSourcesService extends BaseService { RestBusinessCode.ReuseRegionIsNull.getValue()); } } + + /** + * 设置配置全量同步状态 + * + * @param value + */ + public void setAllConfigSyncStatus(String value) { + JedisClusterUtils.set("allConfigSyncStatus", value, Constants.CONFIGSYNCLOCKTIME.intValue()); + } + + /** + * 获取配置全量同步状态 + * + * @return + */ + public String getAllConfigSyncStatus() { + return JedisClusterUtils.get("allConfigSyncStatus"); + } + + /** + * 将界面发过来的数据存储到rediscluster中 + * + * @param key + * @param value + */ + public void setRedisClusterKey(String key, String value) { + JedisClusterUtils.set(key, value, 0); + } + + /** + * 将所有业务的配置key记录下来,方便读取 + * + * @param value + */ + public void setAllServiceKey(String value) { + JedisCluster resource = JedisClusterUtils.getResource(); + resource.append("allConfigSyncKey", value + ";"); + } + + public Map getAllConfig() { + Jedis resource = JedisUtils.getResource(0); + Set effectiveSet = new HashSet<>(); + Set obsoleteSet = new HashSet<>(); + for (int i = 2; i < 6; i++) { + resource.select(i); + effectiveSet.addAll(resource.keys("EFFECTIVE_RULE:*_COMPILE*")); + obsoleteSet.addAll(resource.keys("OBSOLETE_RULE:*_COMPILE*")); + } + Map map = new HashMap<>(); + map.put("effectiveMaatKeys", effectiveSet.size()); + map.put("obsoleteMaatKeys", obsoleteSet.size()); + JedisUtils.returnBrokenResource(resource); + return map; + } + + public Map getAllConfigByScan() { + Jedis resource = JedisUtils.getResource(0); + Set effectiveSet = new HashSet<>(); + Set obsoleteSet = new HashSet<>(); + for (int i = 2; i < 6; i++) { + resource.select(i); + effectiveSet.addAll(getKeyByScan("EFFECTIVE_RULE:*_COMPILE*", resource)); + obsoleteSet.addAll(getKeyByScan("OBSOLETE_RULE:*_COMPILE*", resource)); + } + Map map = new HashMap<>(); + map.put("effectiveMaat", effectiveSet.size()); + map.put("obsoleteMaat", obsoleteSet.size()); + JedisUtils.returnBrokenResource(resource); + return map; + } + + public List getKeyByScan(String pattern, Jedis resource) { + List list = new ArrayList<>(); + int count = 1000; + String cursor = ScanParams.SCAN_POINTER_START; + ScanParams scanParams = new ScanParams(); + scanParams.count(count); + scanParams.match(pattern); + + do { + ScanResult scanResult = resource.scan(cursor, scanParams); + list.addAll(scanResult.getResult()); + cursor = scanResult.getStringCursor(); + } while (!"0".equals(cursor)); + return list; + } } diff --git a/src/main/java/com/nis/web/task/SyncAllConfigTask.java b/src/main/java/com/nis/web/task/SyncAllConfigTask.java new file mode 100644 index 0000000..7b69bf8 --- /dev/null +++ b/src/main/java/com/nis/web/task/SyncAllConfigTask.java @@ -0,0 +1,172 @@ +package com.nis.web.task; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.UUID; + +import org.apache.commons.codec.digest.DigestUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.PropertySource; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import com.nis.domain.restful.ConfigSource; +import com.nis.restful.RestBusinessCode; +import com.nis.restful.ServiceRuntimeException; +import com.nis.util.Constants; +import com.nis.util.ExceptionUtil; +import com.nis.web.service.AuditLogThread; +import com.nis.web.service.restful.ConfigSourcesService; +import com.zdjizhi.utils.JsonMapper; + +import redis.clients.jedis.JedisCluster; + +@Component +@PropertySource(value = { "classpath:nis.properties" }) +public class SyncAllConfigTask { + private static Logger logger = LoggerFactory.getLogger(SyncAllConfigTask.class); + + @Autowired + private JedisCluster jedisCluster; + @Autowired + protected ConfigSourcesService configSourcesService; +// @Scheduled(cron = "${syncUiAndServiceConfigCron}") + public void syncRedisToCluster() { + String requestId = UUID.randomUUID().toString(); + try { +// + if (lock(requestId)) {// 避免集群环境下同一秒钟所有的机器都执行这个定时任务 + String allConfigSyncStatus = jedisCluster.get("allConfigSyncStatus"); + if (allConfigSyncStatus != null) {// 配置初始化完成 + if (allConfigSyncStatus.trim().equals("1")) { + logger.warn(""); + // 设置配置同步状态为正在进行 + configSourcesService.setAllConfigSyncStatus("2"); + logger.warn("开始执行配置全量导入操作,将allConfigSyncStatus值设置为2正在进行导入操作"); + Map maatMap = new HashMap<>(); + Map unMaatMap = new HashMap<>(); + String allConfigSyncKey = jedisCluster.get("allConfigSyncKey"); + if (allConfigSyncKey != null && !allConfigSyncKey.trim().equals("")) { + String[] split = org.apache.commons.lang.StringUtils.split(allConfigSyncKey, ";"); + for (String key : split) { + String val = jedisCluster.get(key); + String md5 = DigestUtils.md5Hex(val); + if (key.startsWith("UNMAAT")) { + unMaatMap.put(md5, val); + } else if (key.startsWith("MAAT")) { + maatMap.put(md5, val); + } + } + addConfigToRedis(maatMap,true); + addConfigToRedis(unMaatMap,false); + logger.warn("执行配置全量导入成功,将allConfigSyncStatus值设置为3,导入成功"); + // 设置配置同步状态为写redis成功 + configSourcesService.setAllConfigSyncStatus("3"); + + // 删除存储全量配置key的关系key + jedisCluster.del("allConfigSyncKey"); + for (String key : split) { + jedisCluster.del(key); + } + logger.warn("删除allConfigSyncKey,及其中的内容成功"); + } + } else { + logger.info( + "集群中allConfigSyncStatus的值是{}[开始:0(界面下发同步状态),初始化:1(配置接收完成状态),进行中:2(服务写redis),已完成:3(服务写redis完毕),失败:-1(服务写redis失败)],暂不执行全量配置同步操作", + allConfigSyncStatus); + } + } else { + logger.info("未从集群中获取到allConfigSyncStatus的值,暂不执行全量配置同步操作"); + } + + } else { + logger.info("没有从rediscluster中获取到allConfigSyncDistributedLock分布式锁,暂时不执行数据同步!"); + } + } catch (Exception e) { + logger.error("同步界面配置到redis中失败,失败原因:{}", ExceptionUtil.getExceptionMsg(e)); + // 设置配置同步状态为写redis失败 + configSourcesService.setAllConfigSyncStatus("-1"); + logger.error("执行配置全量导入失败,将allConfigSyncStatus值设置为-1,导入失败"); + } finally { + unlock(requestId); + closeConn(); + } + } + + public void addConfigToRedis(Map map, boolean isMaat) throws Exception { + long time = System.currentTimeMillis(); + StringBuffer sb = new StringBuffer(); + if (isMaat) { + for (Entry entry : map.entrySet()) { + ConfigSource configSource = new JsonMapper().fromJson(entry.getValue(), ConfigSource.class); + configSourcesService.saveMaatConfig(new AuditLogThread(), time, configSource.getConfigCompileList(), + sb); + } + } else { + for (Entry entry : map.entrySet()) { + String value = entry.getValue(); + configSourcesService.saveCommonSources(new AuditLogThread(), time, value); + } + + } + + } + + public void closeConn() { + if (jedisCluster != null) { + try { + jedisCluster.close(); + } catch (Exception e) { + throw new ServiceRuntimeException("释放redis-cluster连接失败", + RestBusinessCode.CannotConnectionRedis.getValue()); + } + } + } + +// 设置成功返回的结果OK + private static final String LOCK_SUCCESS = "OK"; +// NX -- Only set the key if it does not already exist. XX -- Only set the key +// if it already exist + private static final String SET_IF_NOT_EXIST = "NX"; +// 失效单位秒(EX)还是毫秒(PX) + private static final String SET_WITH_EXPIRE_TIME = "EX"; + private static final Long UNLOCK_SUCCESS = 1L; + + /** + * 尝试获取分布式锁,如果没有key就set,有key就不操作 + * + * @param requestId 请求标识(UUID.randomUUID().toString()),正常情况下是谁加的锁,谁去解锁不能a加的锁,b去解锁 + * @return 是否获取成功 + */ + public Boolean lock(String requestId) { + String key = "allConfigSyncDistributedLock"; + String var1 = jedisCluster.set(key, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, + Constants.CONFIGSYNCLOCKTIME); + if (LOCK_SUCCESS.equals(var1)) { + return true; + } + return false; + } + + /** + * 解除redis分布式锁 + * + * @param requestId + */ + protected boolean unlock(String requestId) { + String key = "allConfigSyncDistributedLock"; + // 这个字符串是个lua脚本，代表的意思是如果根据key拿到的value跟传入的value相同就执行del，否则就返回0【保证安全性】 + String luaScript = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then return redis.call(\"del\",KEYS[1]) else return 0 end"; + // 这个命令就是去执行lua脚本，KEYS的集合就是第二个参数，ARGV的集合就是第三参数【保证解锁的原子操作】 + Object var2 = jedisCluster.eval(luaScript, Collections.singletonList(key), + Collections.singletonList(requestId)); + if (UNLOCK_SUCCESS == var2) { + return true; + } + return false; + } +} diff --git a/src/main/resources/nis.properties b/src/main/resources/nis.properties index 04aeaab..b34bc84 100644 --- a/src/main/resources/nis.properties +++ b/src/main/resources/nis.properties @@ -213,7 +213,13 @@ fileProtocol=redis:// #是否开启日志查询count和last功能 isOpenLogCountAndLast=true -#redis分布式锁超时时间,默认五分钟,3000秒 -redisLockTime=3000 +#redis分布式锁超时时间,默认五分钟,300秒 +redisLockTime=300 #获取redis分布式锁失败后的尝试获取锁的次数,每次失败暂停一秒钟后再次尝试 -redisRetryNum=5 \ No newline at end of file +redisRetryNum=5 + +#定时检测界面和服务的配置是否需要同步的定时任务间隔 +syncUiAndServiceConfigCron=0/10 * * * * ? +#全量数据同步分布式锁时间,默认两个小时 +configSyncLockTime=7200 +