测试

+package com.yoho.unions.common.redis;
+
+import com.alibaba.fastjson.JSON;
+import com.yoho.core.redis.cluster.annotation.Redis;
+import com.yoho.core.redis.cluster.operations.nosync.YHRedisTemplate;
+import com.yoho.core.redis.cluster.operations.nosync.YHValueOperations;
+import com.yoho.core.redis.cluster.operations.serializer.RedisKeyBuilder;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by mingdan.ge on 2018/9/25.
+ */
+@Component
+public class GlobalRedisValueCache {
+    private final static Logger log = LoggerFactory.getLogger(RedisValueCache.class);
+
+    @Redis("yohoGlobalRedis")
+    YHValueOperations yhValueOperationsGlobal;
+
+    @Redis("yohoGlobalRedis")
+    YHRedisTemplate yHRedisTemplate;
+
+    public boolean getLock(String key,int time,TimeUnit timeUnit) {
+        try {
+            boolean lock = setIfAbsent(key, "lock");
+            if (!lock) {
+                return false;
+            }
+            longExpire(key, time, timeUnit);
+            return true;
+        } catch (Exception e) {
+            log.error("getLock: {} error, err is {}", key, e.getMessage());
+            try {
+                delete(key);
+            } catch (Exception e1) {
+                log.error(key, e1);
+            }
+            return false;
+        }
+    }
+
+
+
+    /**
+     * 设置
+     * @param key
+     * @param value
+     */
+    public <T> void set(String key, T value ,long timeout, TimeUnit unit) {
+        try {
+            String v = RedisUtil.value2String(value);
+            if (v == null) {
+                return;
+            }
+            RedisKeyBuilder redisKeyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
+            yhValueOperationsGlobal.set(redisKeyBuilder, v, timeout, unit);
+        }catch (Exception e){
+            log.warn("Redis exception. value redis set . key is {}, value is {}, error msg is {}", key, value, e.getMessage());
+        }
+
+    }
+
+    /**
+     * 设置
+     * @param key
+     * @param value
+     */
+    public <T> boolean setIfAbsent(String key, T value) {
+        String v = RedisUtil.value2String(value);
+        if (v == null) {
+            return false;
+        }
+        RedisKeyBuilder redisKeyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
+        return yhValueOperationsGlobal.setIfAbsent(redisKeyBuilder, v);
+    }
+
+
+    /**
+     * 获取值
+     * @param key
+     * @param clazz
+     * @return
+     */
+    public <T> T get(Object key, Class<T> clazz) {
+        try {
+            RedisKeyBuilder redisKeyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
+            String v = yhValueOperationsGlobal.get(redisKeyBuilder);
+            if (v == null) {
+                return null;
+            }
+            return RedisUtil.string2Value(v, clazz);
+        }catch (Exception e){
+            log.warn("get redis value operation failed. key is {},  error msg is {}", key, e.getMessage());
+        }
+        return null;
+    }
+
+    /**
+     * increment
+     * @param key
+     * @param delta
+     * @return
+     */
+    public Long increment(String key, long delta) {
+        RedisKeyBuilder redisKeyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
+        return yhValueOperationsGlobal.increment(redisKeyBuilder, delta);
+    }
+
+    public static <T> T string2Value(String value, Class<T> clazz) {
+        if (StringUtils.isEmpty(value)) {
+            return null;
+        }
+        if (clazz.getName().equalsIgnoreCase("java.lang.String")) {
+            return (T) value;
+        }
+        return (T) JSON.parseObject(value, clazz);
+    }
+
+
+    /**
+     * longExpire
+     * @param key
+     * @param time
+     * @param unit
+     * @return
+     */
+    public boolean longExpire(String key,long time,TimeUnit unit){
+        RedisKeyBuilder redisKeyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
+        return this.yHRedisTemplate.longExpire(redisKeyBuilder, time, unit);
+    }
+
+    /**
+     * delete
+     * @param key
+     * @return
+     */
+    public void delete(String key) {
+        RedisKeyBuilder redisKeyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
+        yHRedisTemplate.delete(redisKeyBuilder);
+    }
+
+}

@@ -2,7 +2,7 @@ package com.yoho.unions.server.task;
 
 import com.yoho.error.utils.LocalhostIpFetcher;
 import com.yoho.service.model.union.bo.CpsMessageTaskBo;
-import com.yoho.unions.common.redis.RedisValueCache;
+import com.yoho.unions.common.redis.GlobalRedisValueCache;
 import com.yoho.unions.common.utils.DateUtil;
 import com.yoho.unions.server.service.IUnionShareService;
 import org.slf4j.Logger;
@@ -26,7 +26,7 @@ public class CpsOrdersSumMessageTask {
     IUnionShareService unionShareService;
 
     @Resource
-    RedisValueCache redisValueCache;
+    GlobalRedisValueCache redisValueCache;
 
     private static String TASK_KEY = "yh:union:share:CpsOrdersSumMessageTask:";
 
@@ -34,28 +34,35 @@ public class CpsOrdersSumMessageTask {
 //    @Scheduled(cron = "0 0 10 * * ?")
     public void run() {
         //绑定已处理状态
-        CpsMessageTaskBo bo = new CpsMessageTaskBo();
-        bo.setIp(LocalhostIpFetcher.fetchLocalIP());
-        bo.setTime(Long.toString(System.currentTimeMillis()));
-        CpsMessageTaskBo cacheBo=redisValueCache.get(TASK_KEY+DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
-        if (cacheBo != null) {
-            logger.info("other get task,it is {}.this is {}.",cacheBo,bo);
-        }
-        redisValueCache.set(TASK_KEY+DateUtil.getToday("yyyyMMdd"),bo,2, TimeUnit.DAYS);
-        try {
-            Thread.sleep(1000);
-        } catch (InterruptedException e) {
-            logger.warn("thread sleep error,e is {}",e);
-        }
-        cacheBo=redisValueCache.get(TASK_KEY+DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
-        if (cacheBo == null) {
-            logger.error("redis error.");
-            return;
-        }
-        if (!cacheBo.getIp().equals(bo.getIp()) || !cacheBo.getTime().equals(bo.getTime())) {
-            logger.info("The competition failure,other get task,it is {}.this is {}.",cacheBo,bo);
-            return;
+//        CpsMessageTaskBo bo = new CpsMessageTaskBo();
+//        bo.setIp(LocalhostIpFetcher.fetchLocalIP());
+//        bo.setTime(Long.toString(System.currentTimeMillis()));
+//        CpsMessageTaskBo cacheBo=redisValueCache.get(TASK_KEY+DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
+//        if (cacheBo != null) {
+//            logger.info("other get task,it is {}.this is {}.",cacheBo,bo);
+//        }
+//        redisValueCache.set(TASK_KEY+DateUtil.getToday("yyyyMMdd"),bo,2, TimeUnit.DAYS);
+//        try {
+//            Thread.sleep(1000);
+//        } catch (InterruptedException e) {
+//            logger.warn("thread sleep error,e is {}",e);
+//        }
+//        cacheBo=redisValueCache.get(TASK_KEY+DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
+//        if (cacheBo == null) {
+//            logger.error("redis error.");
+//            return;
+//        }
+//        if (!cacheBo.getIp().equals(bo.getIp()) || !cacheBo.getTime().equals(bo.getTime())) {
+//            logger.info("The competition failure,other get task,it is {}.this is {}.",cacheBo,bo);
+//            return;
+//        }
+
+        String key = TASK_KEY + DateUtil.getToday("yyyyMMdd");
+        boolean lock = redisValueCache.getLock(key,2, TimeUnit.DAYS);
+        if (!lock) {
+            logger.info("The competition failure,other get task.");
         }
+        logger.info("get this task,ip is {}",LocalhostIpFetcher.fetchLocalIP());
 
         //昨日
         int startTime=DateUtil.getLastDayInt(1);

@@ -2,7 +2,7 @@ package com.yoho.unions.server.task;
 
 import com.yoho.error.utils.LocalhostIpFetcher;
 import com.yoho.service.model.union.bo.CpsMessageTaskBo;
-import com.yoho.unions.common.redis.RedisValueCache;
+import com.yoho.unions.common.redis.GlobalRedisValueCache;
 import com.yoho.unions.common.utils.DateUtil;
 import com.yoho.unions.server.service.IUnionShareService;
 import org.slf4j.Logger;
@@ -25,7 +25,7 @@ public class CpsSettlementMessageTask {
     @Autowired
     IUnionShareService unionShareService;
     @Resource
-    RedisValueCache redisValueCache;
+    GlobalRedisValueCache redisValueCache;
 
     private static String SETTLEMENT_TASK_KEY = "yh:union:share:CpsSettlementMessageTask:";
 
@@ -34,29 +34,35 @@ public class CpsSettlementMessageTask {
 //    @Scheduled(cron = "0 0 10 20,28 * ?")
     public void run(){
         //竞争任务处理者
-        CpsMessageTaskBo bo = new CpsMessageTaskBo();
-        bo.setIp(LocalhostIpFetcher.fetchLocalIP());
-        bo.setTime(Long.toString(System.currentTimeMillis()));
-        CpsMessageTaskBo cacheBo=redisValueCache.get(SETTLEMENT_TASK_KEY+ DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
-        if (cacheBo != null) {
-            logger.info("other get task,it is {}.this is {}.",cacheBo,bo);
-        }
-        redisValueCache.set(SETTLEMENT_TASK_KEY+DateUtil.getToday("yyyyMMdd"),bo,2, TimeUnit.DAYS);
-        try {
-            Thread.sleep(1000);
-        } catch (InterruptedException e) {
-            logger.warn("thread sleep error,e is {}",e);
-        }
-        cacheBo=redisValueCache.get(SETTLEMENT_TASK_KEY+DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
-        if (cacheBo == null) {
-            logger.error("redis error.");
-            return;
-        }
-        if (!cacheBo.getIp().equals(bo.getIp()) || !cacheBo.getTime().equals(bo.getTime())) {
-            logger.info("The competition failure,other get task,it is {}.this is {}.",cacheBo,bo);
-            return;
-        }
+//        CpsMessageTaskBo bo = new CpsMessageTaskBo();
+//        bo.setIp(LocalhostIpFetcher.fetchLocalIP());
+//        bo.setTime(Long.toString(System.currentTimeMillis()));
+//        CpsMessageTaskBo cacheBo=redisValueCache.get(SETTLEMENT_TASK_KEY+ DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
+//        if (cacheBo != null) {
+//            logger.info("other get task,it is {}.this is {}.",cacheBo,bo);
+//        }
+//        redisValueCache.set(SETTLEMENT_TASK_KEY+DateUtil.getToday("yyyyMMdd"),bo,2, TimeUnit.DAYS);
+//        try {
+//            Thread.sleep(1000);
+//        } catch (InterruptedException e) {
+//            logger.warn("thread sleep error,e is {}",e);
+//        }
+//        cacheBo=redisValueCache.get(SETTLEMENT_TASK_KEY+DateUtil.getToday("yyyyMMdd"), CpsMessageTaskBo.class);
+//        if (cacheBo == null) {
+//            logger.error("redis error.");
+//            return;
+//        }
+//        if (!cacheBo.getIp().equals(bo.getIp()) || !cacheBo.getTime().equals(bo.getTime())) {
+//            logger.info("The competition failure,other get task,it is {}.this is {}.",cacheBo,bo);
+//            return;
+//        }
 
+        String key = SETTLEMENT_TASK_KEY + DateUtil.getToday("yyyyMMdd");
+        boolean lock = redisValueCache.getLock(key,2, TimeUnit.DAYS);
+        if (!lock) {
+            logger.info("The competition failure,other get task.");
+        }
+        logger.info("get this task,ip is {}",LocalhostIpFetcher.fetchLocalIP());
         //今天更新的数据
         int startTime=DateUtil.getLastDayInt(0);
         int endTime= DateUtil.getLastDayInt(-1);

@@ -2,4 +2,9 @@ redis:
     yohoNoSyncRedis :
            servers:
              - ${redis.union.readonly.proxy.address}:${redis.union.readonly.proxy.port}
-           auth:  ${redis.union.readonly.proxy.auth}
\ No newline at end of file
+           auth:  ${redis.union.readonly.proxy.auth}
+
+    yohoGlobalRedis :
+               servers:
+                 - ${redis.global.address}
+               auth: ${redis.global.auth}
\ No newline at end of file