release- 积分日志单线程运行

luoo_user/src/main/java/com/luoo/user/config/RabbitMQConfig.java

@@ -25,12 +25,4 @@ public class RabbitMQConfig {
         return factory;
     }
 
-    @Bean
-    public SimpleRabbitListenerContainerFactory pointLogFactory(ConnectionFactory connectionFactory) {
-        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
-        factory.setConnectionFactory(connectionFactory);
-        factory.setMessageConverter(new Jackson2JsonMessageConverter());
-        factory.setPrefetchCount(1);
-        return factory;
-    }
 }

luoo_user/src/main/java/com/luoo/user/listener/PointLogListener.java

@@ -3,9 +3,11 @@ package com.luoo.user.listener;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.luoo.user.pojo.UserPointLog;
 import com.luoo.user.service.UserPointLogService;
+import com.luoo.user.util.DistributedLock;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -18,13 +20,19 @@ import java.io.IOException;
  * @create: 2024-07-24 13:05
  **/
 @Component
-@RabbitListener(queues = "pointLog", containerFactory = "pointLogFactory")
+@RabbitListener(queues = "pointLog")
 @Slf4j
 public class PointLogListener {
 
   @Autowired
   private UserPointLogService userPointLogService;
 
+  @Autowired
+  private DistributedLock distributedLock;
+
+  @Autowired
+  private RabbitTemplate rabbitTemplate;
+
   @RabbitHandler
   public void executePointLog(String json) {
     log.info("userPointLog:{}", json);
@@ -33,7 +41,24 @@ public class PointLogListener {
 
     try {
       UserPointLog userPointLog = objectMapper.readValue(json, UserPointLog.class);
+      String userId = userPointLog.getUserId();
+      // 对userId进行redis分布式锁
+      // springboot 写一段setnx的分布式锁
+
+      String key = "pointLog:" + userId;
+
+      if (distributedLock.tryLock(key, "1")) {
+        try {
+          // 执行需要加锁的业务逻辑
           userPointLogService.add(userPointLog);
+        } finally {
+          // 释放锁
+          distributedLock.unlock(key);
+        }
+      } else {
+        // 获取锁失败，处理相应逻辑
+        rabbitTemplate.convertAndSend("pointLog", json);
+      }
     } catch (IOException e) {
       e.printStackTrace();
     }

luoo_user/src/main/java/com/luoo/user/util/DistributedLock.java

@@ -0,0 +1,21 @@
+package com.luoo.user.util;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+public class DistributedLock {
+
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+
+    public boolean tryLock(String key, String value) {
+        Boolean success = redisTemplate.opsForValue().setIfAbsent(key, value);
+        return success != null && success;
+    }
+
+    public void unlock(String key) {
+        redisTemplate.delete(key);
+    }
+}