新增redisson锁和redisson锁切面，保证分布式锁，分布式id

mall-common/src/main/java/com/suisung/mall/common/annotation/DistributedLock.java

@@ -0,0 +1,39 @@
+package com.suisung.mall.common.annotation;
+
+import java.lang.annotation.*;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 分布式锁注解
+ */
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface DistributedLock {
+    /**
+     * 锁的key，支持SpEL表达式,下面是示例
+     * key = "'ACCOUNT_LOCK:' + #batchSize", // 使用SpEL表达式，锁key包含参数
+     * public List<Integer> getBatchUserAccountBaseId(int batchSize)
+     */
+    String key();
+
+    /**
+     * 等待时间（秒），默认0-不等待
+     */
+    long waitTime() default 0;
+
+    /**
+     * 锁持有时间（秒），默认30秒，-1表示使用看门狗机制
+     */
+    long leaseTime() default 30;
+
+    /**
+     * 时间单位，默认秒
+     */
+    TimeUnit timeUnit() default TimeUnit.SECONDS;
+
+    /**
+     * 获取锁失败时的错误消息
+     */
+    String errorMsg() default "系统繁忙，请稍后再试";
+}

mall-shop/src/main/java/com/suisung/mall/shop/Aspect/DistributedLockAspect.java

@@ -0,0 +1,110 @@
+package com.suisung.mall.shop.Aspect;
+
+import com.suisung.mall.common.annotation.DistributedLock;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.reflect.MethodSignature;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.DefaultParameterNameDiscoverer;
+import org.springframework.core.ParameterNameDiscoverer;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.stereotype.Component;
+
+import java.lang.reflect.Method;
+
+/**
+ * 分布式锁切面
+ */
+@Slf4j
+@Aspect
+@Component
+public class DistributedLockAspect {
+    @Autowired
+    private RedissonClient redissonClient;
+
+    private final ExpressionParser expressionParser = new SpelExpressionParser();
+    private final ParameterNameDiscoverer parameterNameDiscoverer = new DefaultParameterNameDiscoverer();
+
+    @Around("@annotation(distributedLock)")
+    public Object around(ProceedingJoinPoint joinPoint, DistributedLock distributedLock) throws Throwable {
+        // 解析锁的key，支持SpEL表达式
+        String lockKey = parseLockKey(joinPoint, distributedLock);
+
+        RLock lock = redissonClient.getLock(lockKey);
+        boolean isLocked = false;
+
+        try {
+            // 尝试获取锁
+            if (distributedLock.leaseTime() == -1) {
+                // 使用看门狗机制
+                isLocked = lock.tryLock(distributedLock.waitTime(), distributedLock.timeUnit());
+            } else {
+                isLocked = lock.tryLock(distributedLock.waitTime(), distributedLock.leaseTime(), distributedLock.timeUnit());
+            }
+
+            if (!isLocked) {
+                log.warn("获取分布式锁失败，lockKey: {}", lockKey);
+                throw new RuntimeException(distributedLock.errorMsg());
+            }
+
+            log.debug("成功获取分布式锁，lockKey: {}", lockKey);
+            // 执行原方法
+            return joinPoint.proceed();
+
+        } finally {
+            // 释放锁
+            if (isLocked && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+                log.debug("释放分布式锁，lockKey: {}", lockKey);
+            }
+        }
+    }
+
+    /**
+     * 解析锁的key，支持SpEL表达式
+     */
+    private String parseLockKey(ProceedingJoinPoint joinPoint, DistributedLock distributedLock) {
+        String keyExpression = distributedLock.key();
+
+        // 如果key不包含SpEL表达式，直接返回
+        if (!keyExpression.contains("#")) {
+            return keyExpression;
+        }
+
+        try {
+            MethodSignature signature = (MethodSignature) joinPoint.getSignature();
+            Method method = signature.getMethod();
+            Object[] args = joinPoint.getArgs();
+            String[] parameterNames = parameterNameDiscoverer.getParameterNames(method);
+
+            EvaluationContext context = new StandardEvaluationContext();
+            if (parameterNames != null) {
+                for (int i = 0; i < parameterNames.length; i++) {
+                    context.setVariable(parameterNames[i], args[i]);
+                }
+            }
+
+            // 设置一些常用变量
+            context.setVariable("methodName", method.getName());
+            context.setVariable("className", method.getDeclaringClass().getSimpleName());
+
+            Expression expression = expressionParser.parseExpression(keyExpression);
+            String result = expression.getValue(context, String.class);
+
+            return StringUtils.isNotBlank(result) ? result : keyExpression;
+
+        } catch (Exception e) {
+            log.warn("解析锁key表达式失败，使用原表达式: {}, error: {}", keyExpression, e.getMessage());
+            return keyExpression;
+        }
+    }
+}

mall-shop/src/main/java/com/suisung/mall/shop/config/RedissionConfig.java

@@ -37,7 +37,13 @@ public class RedissionConfig {
         config.useSingleServer().setAddress("redis://" + host + ":" + port);
         config.useSingleServer().setDatabase(database);
         config.useSingleServer().setPassword(password);
-
+        config.useSingleServer().setTimeout(1000*120);// 命令超时时间（毫秒）
+        config.useSingleServer().setRetryAttempts(5);// 重试次数
+        config.useSingleServer().setRetryInterval(2000); // 重试间隔（毫秒）
+        // 设置netty线程数
+        config.setNettyThreads(32);
+        // 增加处理线程数
+        config.setThreads(16);
         //2、根据Config创建出RedissonClient实例
         return Redisson.create(config);
     }

mall-shop/src/main/java/com/suisung/mall/shop/number/service/impl/ShopNumberSeqServiceImpl.java

@@ -2,6 +2,8 @@ package com.suisung.mall.shop.number.service.impl;
 
 import cn.hutool.core.date.DateUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.suisung.mall.common.annotation.DistributedLock;
+import com.suisung.mall.common.exception.ApiException;
 import com.suisung.mall.common.feignService.AccountService;
 import com.suisung.mall.common.modules.base.ShopBaseProductSpec;
 import com.suisung.mall.common.modules.library.LibraryProduct;
@@ -19,6 +21,8 @@ import com.suisung.mall.shop.product.service.ShopProductBaseService;
 import com.suisung.mall.shop.product.service.ShopProductItemService;
 import com.suisung.mall.shop.product.service.ShopProductSpecItemService;
 import com.suisung.mall.shop.sync.keymanage.RedisKey;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
@@ -29,6 +33,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.LongStream;
@@ -71,6 +76,9 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
     @Lazy
     private ShopProductItemService shopProductItemService;
 
+    @Autowired
+    private RedissonClient redissonClient;
+
 
     public static void main(String[] args) {
         System.out.printf(IntStream.rangeClosed(1, 1).boxed().collect(Collectors.toList()).toString());
@@ -85,8 +93,13 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
      */
     @Override
     @Transactional(propagation = Propagation.NOT_SUPPORTED)
+    @DistributedLock(
+            key = "CREATENEXTSEQ_LOCK", // 锁的key
+            waitTime = 3,                       // 等待3秒
+            leaseTime = 10,                     // 锁持有10秒
+            errorMsg = "生成ID繁忙，请稍后重试" // 自定义错误消息
+    )
     public synchronized String createNextSeq(String prefix) {
-
         String ymd = DateUtil.format(new Date(), "yyyyMMdd");
         String id = String.format("%s_%s_", prefix, ymd);
         ShopNumberSeq shopNumberSeq = this.baseMapper.selectById(id);
@@ -148,25 +161,6 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
         return null;
     }
 
-    /**
-     * 校验同步数据是否同步完成，就是两个库的主键是否一致
-     *
-     * @return
-     */
-    private void checkPrimaryKey() {
-        boolean flag = shopNumberSeqMapper.findNumberFromShopBaseAndItem(new ShopNumberSeq()).size() > 1;
-        while (!flag) {
-            try {
-                Thread.sleep(3600);
-            } catch (InterruptedException e) {
-                log.error("checkPrimaryKey枷锁失败--" + e.getMessage());
-                break;
-            }
-            flag = shopNumberSeqMapper.findNumberFromShopBaseAndItem(new ShopNumberSeq()).size() > 1;
-        }
-
-    }
-
     /**
      * 批量获取id
      *
@@ -175,11 +169,18 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
      * @return
      */
     @Transactional(propagation = Propagation.NOT_SUPPORTED)
+    @DistributedLock(
+            key = "'CREATENEXTSEQ_LOCK:'+#seqName", // 锁的key
+            waitTime = 3,                       // 等待3秒
+            leaseTime = 10,                     // 锁持有10秒
+            errorMsg = "生成ID繁忙，请稍后重试" // 自定义错误消息
+    )
     public synchronized List<Long> batchCreateNextNo(String seqName, int batchSize) {
         if (batchSize <= 0) {
             return Collections.emptyList();
         }
         long number = 0;
+
         if (null == redisService.get(String.format(CACHE_PREFIX, seqName))) {
             syncPrimaryKey();
             // 1. 获取当前序列值
@@ -200,7 +201,6 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
             number = numberCache;
 
         }
-
         // 2. 计算新值范围
         long start = number;
         long end = start + batchSize - 1;
@@ -269,8 +269,20 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
      * @return
      */
     @Override
-    public synchronized List<Integer> getBatchSpecId(int batchSize) {
+    public List<Integer> getBatchSpecId(int batchSize) {
+        // 定义锁的key，这个key在所有服务实例中必须一致
+        String lockKey = "LOCK:" + RedisKey.STOREDATASPECID;
+        // 2. 获取分布式锁对象
+        RLock lock = redissonClient.getLock(lockKey);
+        boolean isLocked = false;
         int start = 0;
+        try {
+             isLocked =  lock.tryLock(2, 30, TimeUnit.SECONDS);
+            if (!isLocked) {
+                // 获取锁失败，可以根据业务逻辑进行重试或抛出异常
+                throw new RuntimeException("系统繁忙，请稍后再试");
+            }
+            log.info("成功获得锁：{}",lockKey);
             if (null != redisService.get(RedisKey.STOREDATASPECID)) {
                 start = (Integer) redisService.get(RedisKey.STOREDATASPECID);
                 redisService.set(RedisKey.STOREDATASPECID, start + batchSize);
@@ -288,6 +300,16 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
                 return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
             }
             return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
+        } catch (Exception e) {
+            throw new ApiException(e);
+        }finally {
+            // 5. 最终检查并释放锁，确保锁一定被释放
+            if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+            log.info("成功释放锁：{}",lockKey);
+        }
+
     }
 
     /**
@@ -297,8 +319,20 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
      * @return
      */
     @Override
-    public synchronized List<Integer> getBatchSpecItemId(int batchSize) {
+    public  List<Integer> getBatchSpecItemId(int batchSize) {
+        // 定义锁的key，这个key在所有服务实例中必须一致
+        String lockKey = "LOCK:" + RedisKey.STOREDATASPECITEMID;
+        // 2. 获取分布式锁对象
+        RLock lock = redissonClient.getLock(lockKey);
+        boolean isLocked = false;
         int start = 0;
+        try {
+             isLocked=lock.tryLock(2,30,TimeUnit.SECONDS);
+            if (!isLocked) {
+                // 获取锁失败，可以根据业务逻辑进行重试或抛出异常
+                throw new ApiException("系统繁忙，请稍后再试");
+            }
+            log.info("成功获得锁：{}",lockKey);
             if (null != redisService.get(RedisKey.STOREDATASPECITEMID)) {
                 start = (Integer) redisService.get(RedisKey.STOREDATASPECITEMID);
                 redisService.set(RedisKey.STOREDATASPECITEMID, start + batchSize);
@@ -316,6 +350,15 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
                 return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
             }
             return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
+        } catch (InterruptedException e) {
+            throw new ApiException(e);
+        }finally {
+            // 5. 最终检查并释放锁，确保锁一定被释放
+            if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+            log.info("成功释放锁：{}",lockKey);
+        }
     }
 
     /**
@@ -325,7 +368,19 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
      * @return
      */
     @Override
-    public synchronized List<Integer> getBatchUserAccountBaseId(int batchSize) {
+    public List<Integer> getBatchUserAccountBaseId(int batchSize) {
+        // 定义锁的key，这个key在所有服务实例中必须一致
+        String lockKey = "LOCK:" + RedisKey.STOREDATACCOUNTBASEID;
+        // 2. 获取分布式锁对象
+        RLock lock = redissonClient.getLock(lockKey);
+        boolean isLocked = false;
+        try {
+            isLocked = lock.tryLock(2, 10, TimeUnit.SECONDS);
+            if (!isLocked) {
+                // 获取锁失败，可以根据业务逻辑进行重试或抛出异常
+                throw new RuntimeException("系统繁忙，请稍后再试");
+            }
+            log.info("成功获得锁：{}",lockKey);
             int start = 0;
             if (null != redisService.get(RedisKey.STOREDATACCOUNTBASEID)) {
                 start = (Integer) redisService.get(RedisKey.STOREDATACCOUNTBASEID);
@@ -342,6 +397,17 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
                 return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
             }
             return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("getBatchUserAccountBaseId获取锁时被中断", e);
+        }finally {
+            // 5. 最终检查并释放锁，确保锁一定被释放
+            if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+            log.info("成功释放锁：{}",lockKey);
+        }
+
     }
 
 
@@ -352,8 +418,20 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
      * @return
      */
     @Override
-    public synchronized List<Integer> getBatchLibraryProductId(int batchSize) {
+    public  List<Integer> getBatchLibraryProductId(int batchSize) {
+        // 定义锁的key，这个key在所有服务实例中必须一致
+        String lockKey = "LOCK:" + RedisKey.STOREDATALIBRARYID;
+        // 2. 获取分布式锁对象
+        RLock lock = redissonClient.getLock(lockKey);
+        boolean isLocked = false;
+        try {
+            isLocked = lock.tryLock(2, 10, TimeUnit.SECONDS);
             int start = 0;
+            if (!isLocked) {
+                // 获取锁失败，可以根据业务逻辑进行重试或抛出异常
+                throw new RuntimeException("系统繁忙，请稍后再试");
+            }
+            log.info("成功获得锁：{}",lockKey);
             if (null != redisService.get(RedisKey.STOREDATALIBRARYID)) {
                 start = (Integer) redisService.get(RedisKey.STOREDATALIBRARYID);
                 redisService.set(RedisKey.STOREDATALIBRARYID, start + batchSize);
@@ -371,6 +449,16 @@ public class ShopNumberSeqServiceImpl extends BaseServiceImpl<ShopNumberSeqMappe
                 return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
             }
             return IntStream.rangeClosed(start + 1, start + batchSize).boxed().collect(Collectors.toList());
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }finally {
+            // 5. 最终检查并释放锁，确保锁一定被释放
+            if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+            log.info("成功释放锁：{}",lockKey);
+        }
+
     }
 
     @Override

mall-shop/src/main/java/com/suisung/mall/shop/sync/service/impl/SyncThirdDataServiceImpl.java

@@ -24,7 +24,6 @@ import com.google.gson.GsonBuilder;
 import com.qcloud.cos.model.COSObjectSummary;
 import com.suisung.mall.common.adapter.BigDecimalTypeAdapter;
 import com.suisung.mall.common.api.CommonResult;
-import com.suisung.mall.common.api.ResultCode;
 import com.suisung.mall.common.api.StateCode;
 import com.suisung.mall.common.enums.DicEnum;
 import com.suisung.mall.common.exception.ApiException;
@@ -72,10 +71,11 @@ import com.suisung.mall.shop.sync.dto.ActiveShopInfo;
 import com.suisung.mall.shop.sync.dto.BrandModel;
 import com.suisung.mall.shop.sync.keymanage.RedisKey;
 import com.suisung.mall.shop.sync.service.*;
-import com.sun.corba.se.impl.orbutil.concurrent.Sync;
 import io.seata.spring.annotation.GlobalTransactional;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.collections4.ListUtils;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -189,8 +189,11 @@ public class SyncThirdDataServiceImpl extends SyncBaseThirdSxAbstract implements
     @Autowired
     private SyncStoreDataService syncStoreDataService;
 
+
     @Autowired
-    private RedisService redisService;
+    private RedissonClient redissonClient;
+
+
     /**
      * 批量保存商品的分类
      *
@@ -579,13 +582,23 @@ public class SyncThirdDataServiceImpl extends SyncBaseThirdSxAbstract implements
             logger.info("没有商品数据");
             return;
         }
-        String key=RedisKey.STOREDATAGOODBATCHLOCK+":"+storeId;
+
        // redisService.del(key);
-       if(!checkeckIsLock(storeId)){
+        RLock lock =null;
-           logger.info("批量同步商品等待时间异常结束");
+        String lockKey=RedisKey.STOREDATAGOODBATCHLOCK+":"+storeId;
+        try {
+            redissonClient.getKeys().count();
+        }catch (Exception e){
+            logger.info("锁异常");
             return;
         }
-        setLock(storeId,"true");
+         lock = redissonClient.getLock(lockKey); // 锁名称通常与业务关联，如"ORDER_LOCK_123"
+        //尝试三次刚好2分分钟
+        if(!checkeckIsLock(lock)){
+            logger.info("系统繁忙，无法获取锁");
+            return;
+        }
+        try {
                 List<String> newFolders = new ArrayList<>();
                 folders.forEach(page -> {
                     String newfolder = new FileUtils().getSyncTypeFlag(syncType, clientPath) + storeId + FileUtils.pathSeparator + page + FileUtils.pathSeparator;
@@ -724,8 +737,15 @@ public class SyncThirdDataServiceImpl extends SyncBaseThirdSxAbstract implements
                 if(ObjectUtil.isNull(storeDbConfig.getRefreshTime())){
                     syncShopImages(Integer.valueOf(storeId));//同时商品图库数据
                 }
-
+        }catch (Exception e){
-        redisService.del(key);
+            logger.info("执行失败：{}",e.getMessage());
+        }finally {
+            // 最终检查并释放锁，确保锁一定被释放
+            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
+                logger.info("商品同步锁释放");
+                lock.unlock();
+            }
+        }
     }
 
     @Override
@@ -736,37 +756,18 @@ public class SyncThirdDataServiceImpl extends SyncBaseThirdSxAbstract implements
 
     /**
      * 保证单线程执行同步数据
-     * @param storeId
+     * @param lock
      * @return
      */
-    public synchronized boolean  checkeckIsLock(String storeId){
+    public synchronized boolean  checkeckIsLock(RLock lock){
-        logger.info("等待时间开始");
-        String key=RedisKey.STOREDATAGOODBATCHLOCK+":"+storeId;
-        Map isLockMap= (Map) redisService.get(key);
-        if(isLockMap!=null){
-           String result= (String) isLockMap.get("batchGoodIsLock");
-           if("true".equals(result)){
         try {
-                   logger.info("进入等待时间");
+            boolean resul=lock.tryLock(60, 120, TimeUnit.SECONDS);
-                   Thread.sleep(1000 * 60 * 1);//五分钟
+            logger.info("等待获取锁成功");
-                   return checkeckIsLock(storeId);
+            return resul;
         }catch (InterruptedException e){
-                   logger.info("等待异常：{}",e.getMessage());
             return false;
         }
     }
-        }
-        logger.info("等待时间结束");
-        return true;
-    }
-
-    public synchronized void setLock(String storeId,String value){
-        String key=RedisKey.STOREDATAGOODBATCHLOCK+":"+storeId;
-        Map<String,String> map= new HashMap<>();
-        map.put("batchGoodIsLock",value);
-        redisService.set(key,map,600000);
-    }
-
 
     @Override
     public ResponseEntity<Resource> downloadToClient(String primaryKey, String clienVersionName) {