活动注册加锁

xx-activity/xx-activity-server/src/main/java/com/xxfc/platform/activity/biz/ActivityPopularizeBiz.java

@@ -19,14 +19,12 @@ import com.xxfc.platform.universal.feign.MQSenderFeign;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.redisson.api.RLock;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.math.BigDecimal;
 import java.util.Date;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 
 import static com.github.wxiaoqi.security.common.constant.CommonConstants.SYS_FALSE;
 import static com.github.wxiaoqi.security.common.constant.CommonConstants.SYS_TRUE;
@@ -86,13 +84,13 @@ public class ActivityPopularizeBiz extends BaseBiz<ActivityPopularizeMapper, Act
 				}});
 				log.info("活动邀请注册：activityPopularize = {}", activityPopularize.toString());
 				if (activityPopularize != null) {
-					String lockKey = String.format("%s%d%d", registerQueueDTO.getInParamDTO().getActivityCode(), appUserDTO.getInviterAccount(), activityPopularize.getId());
-					RLock rLock = redissonLock.getRLock(lockKey);
-					try {
-						boolean isSuccess = rLock.tryLock(5, 5, TimeUnit.SECONDS);
-						if (isSuccess) {
+//					String lockKey = String.format("%s%d%d", registerQueueDTO.getInParamDTO().getActivityCode(), appUserDTO.getInviterAccount(), activityPopularize.getId());
+//					RLock rLock = redissonLock.getRLock(lockKey);
+//					try {
+//						boolean isSuccess = rLock.tryLock(5, 5, TimeUnit.SECONDS);
+//						if (isSuccess) {
 							log.info("1, {}", System.currentTimeMillis());
-							Thread.sleep(1000);
+							//Thread.sleep(1000);
 							log.info("2, {}", System.currentTimeMillis());
 							List<ActivityPopularizeItem> activityPopularizeItems = activityPopularizeItemBiz.selectByPopularizeId(activityPopularize.getId());
 							ActivityPopularizeItem activityPopularizeItem = activityPopularizeItems.get(activityPopularizeItems.size() - 1);
@@ -206,15 +204,15 @@ public class ActivityPopularizeBiz extends BaseBiz<ActivityPopularizeMapper, Act
 									}}));
 								}
 							}
-						} else {
-							// 获取锁失败
-							log.info("tryLock fail, key = [{}]", lockKey);
-						}
-					} catch (InterruptedException e) {
-						log.error("tryLock fail, key = [{}]", lockKey);
-					} finally {
-						log.info("执行结束， {}", System.currentTimeMillis());
-					}
+//						} else {
+//							// 获取锁失败
+//							log.info("tryLock fail, key = [{}]", lockKey);
+//						}
+//					} catch (InterruptedException e) {
+//						log.error("tryLock fail, key = [{}]", lockKey);
+//					} finally {
+//						log.info("执行结束， {}", System.currentTimeMillis());
+//					}
 				}
 			}
 		}

xx-activity/xx-activity-server/src/main/java/com/xxfc/platform/activity/handler/PopularizeMQHandler.java

@@ -18,8 +18,8 @@ import org.springframework.stereotype.Component;
 
 import java.io.IOException;
 import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.*;
+import java.util.concurrent.locks.StampedLock;
 
 @Component
 @Slf4j
@@ -30,38 +30,50 @@ public class PopularizeMQHandler {
 
     @Autowired
     ActivityUserJoinBiz activityUserJoinBiz;
+    // 同时并发执行的线程数
+    public static int threadTotal = 10;
+
+    final Semaphore semaphore = new Semaphore(threadTotal);
+
+    final CountDownLatch countDownLatch = new CountDownLatch(10);
+
+    private final static StampedLock lock = new StampedLock();
 
     //@RabbitListener(queues = {RabbitActivityConfig.POPULARZIE_REGISTER_QUEUE,RabbitActivityConfig.POPULARZIE_AUTH_QUEUE})
     @RabbitListener(queues = {RabbitActivityConfig.POPULARZIE_0101_QUEUE})
     public void popularizeHandler(Message message, @Headers Map<String, Object> headers, Channel channel) {
         ExecutorService executorService = Executors.newCachedThreadPool();
-        executorService.execute(new Runnable() {
-            @Override
-            public void run() {
-                try {
+        executorService.execute(() -> {
+            try {
+                semaphore.acquire();
                     String messageId = message.getMessageProperties().getMessageId();
                     String msg = new String(message.getBody(), "UTF-8");
                     log.info("接收到的消息：msg = {}, 消息ID是：messageId = {} ", msg, messageId);
-
                     if (StringUtils.isNotBlank(msg)) {
-                        RegisterQueueDTO registerQueueDTO = JSONUtil.toBean(msg, RegisterQueueDTO.class);
-                        activityPopularizeBiz.handleRegister(registerQueueDTO);
+                        long stamp = lock.writeLock();
+                        try {
+                            RegisterQueueDTO registerQueueDTO = JSONUtil.toBean(msg, RegisterQueueDTO.class);
+                            activityPopularizeBiz.handleRegister(registerQueueDTO);
+                        } finally {
+                            lock.unlock(stamp);
+                        }
                     }
 
-                    executorService.shutdown();
                     Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
                     // 手动签收
                     channel.basicAck(deliveryTag, false);
-                } catch (Exception e) {
-                    log.info("接收到的消息失败");
-                    try {
-                        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
-                    } catch (IOException i) {
-                        log.error(e.getMessage(), i);
-                    }
-                    log.error(e.getMessage(), e);
+                semaphore.release();
+            } catch (Exception e) {
+                log.info("接收到的消息失败");
+                try {
+                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
+                } catch (IOException i) {
+                    log.error(e.getMessage(), i);
                 }
+                log.error(e.getMessage(), e);
             }
+            executorService.shutdown();
+            countDownLatch.countDown();
         });
     }
     //新人注册有礼