加锁

mec-biz/src/main/java/com/mec/redisson/RedissonTopicListener.java

@@ -5,6 +5,7 @@ import com.ym.mec.biz.redisson.RedissonMessageService;
 import com.ym.mec.biz.service.ImLiveBroadcastRoomService;
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RBucket;
+import org.redisson.api.RLock;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
@@ -43,28 +44,38 @@ public class RedissonTopicListener implements ApplicationRunner, Ordered {
                     log.error("RedissonMessageService subscribe sleep error", e);
                 }
 
-                // 缓存JoinRoom用户信息到redis
-                RBucket<Object> bucket = redissonClient.getBucket(RedissonMessageService.LIVE_ROOM_MEMBER + message);
-                if (!bucket.isExists()) {
-                   return;
-                }
-                Integer times = (Integer) bucket.get();
-                bucket.delete();
-
-                ImLiveBroadcastRoomVo imLiveBroadcastRoomVo = imLiveBroadcastRoomService.queryRoomInfo(message);
-                if (Objects.isNull(imLiveBroadcastRoomVo)) {
-                    return;
-                }
+                RLock lock = redissonClient.getLock(RedissonMessageService.LIVE_ROOM_MEMBER_LOCK + message);
                 try {
-                    imLiveBroadcastRoomService.setGroupMemberDefinedData(imLiveBroadcastRoomVo,imLiveBroadcastRoomVo.getLookNum(),imLiveBroadcastRoomVo.getTotalLookNum());
-                } catch (Exception e) {
-                    log.error("RedissonMessageService subscribe setGroupMemberDefinedData error", e);
-                    if (times>=3) {
-                        return;
+                    if (lock.tryLock()) {
+                        // 缓存JoinRoom用户信息到redis
+                        RBucket<Object> bucket = redissonClient.getBucket(RedissonMessageService.LIVE_ROOM_MEMBER + message);
+                        if (!bucket.isExists()) {
+                            return;
+                        }
+                        Integer times = (Integer) bucket.get();
+                        bucket.delete();
+
+                        ImLiveBroadcastRoomVo imLiveBroadcastRoomVo = imLiveBroadcastRoomService.queryRoomInfo(message);
+                        if (Objects.isNull(imLiveBroadcastRoomVo)) {
+                            return;
+                        }
+                        try {
+                            imLiveBroadcastRoomService.setGroupMemberDefinedData(imLiveBroadcastRoomVo,imLiveBroadcastRoomVo.getLookNum(),imLiveBroadcastRoomVo.getTotalLookNum());
+                        } catch (Exception e) {
+                            log.error("RedissonMessageService subscribe setGroupMemberDefinedData error", e);
+                            if (times>=3) {
+                                return;
+                            }
+                            bucket.set(times +1, 30, TimeUnit.MINUTES);
+                        }
+                        redissonMessageService.publish(RedissonMessageService.TOPIC_MESSAGE, message);
+                    }
+                } finally {
+                    if (lock.getHoldCount() != 0 && lock.isHeldByCurrentThread()) {
+                        lock.unlock();
                     }
-                    bucket.set(times +1, 30, TimeUnit.MINUTES);
                 }
-                redissonMessageService.publish(RedissonMessageService.TOPIC_MESSAGE, message);
+
 
             });
             log.info("---> RedissonMessageService subscribe success");

mec-biz/src/main/java/com/ym/mec/biz/redisson/RedissonMessageService.java

@@ -17,6 +17,7 @@ public class RedissonMessageService {
     public static final String TOPIC_MESSAGE = "topic:message";
     // 直播在线人数
     public static final String LIVE_ROOM_MEMBER = "delayQueue:liveRoomMember:";
+    public static final String LIVE_ROOM_MEMBER_LOCK = "LOCK:liveRoomMember:";
 
     private final RedissonClient redissonClient;