用户登录和注销时将用户的关注列表和粉丝列表存入数据库

src/main/java/cn/meteor/beyondclouds/common/helper/IRedisHelper.java

@@ -2,6 +2,8 @@ package cn.meteor.beyondclouds.common.helper;
 
 import cn.meteor.beyondclouds.common.exception.RedisOperationException;
 
+import java.util.Set;
+
 /**
  * Redis操作辅助类
  * @author meteor
@@ -96,4 +98,11 @@ public interface IRedisHelper {
      */
     void setRemove(String key, String values);
 
+    /**
+     * 获取set里面的数据
+     * @param key
+     * @return
+     */
+    Set<String> setGet(String key);
+
 }

src/main/java/cn/meteor/beyondclouds/common/helper/impl/RedisHelperImpl.java

@@ -8,6 +8,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.util.Arrays;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -111,4 +112,9 @@ public class RedisHelperImpl implements IRedisHelper {
     public void setRemove(String key, String values) {
         redisTemplate.opsForSet().remove(key, values);
     }
+
+    @Override
+    public Set<String> setGet(String key) {
+        return redisTemplate.opsForSet().members(key);
+    }
 }

src/main/java/cn/meteor/beyondclouds/core/queue/message/UserActionMessage.java

 package cn.meteor.beyondclouds.core.queue.message;
 
 import lombok.Data;
+import lombok.NoArgsConstructor;
 import lombok.ToString;
 
 /**
@@ -8,8 +9,14 @@ import lombok.ToString;
  */
 @Data
 @ToString
+@NoArgsConstructor
 public class UserActionMessage {
 
+    public UserActionMessage(String userId, UserAction action) {
+        this.userId = userId;
+        this.action = action;
+    }
+
     /**
      * 用户ID
      */
@@ -19,4 +26,13 @@ public class UserActionMessage {
      * 操作
      */
     private UserAction action;
+
+    /**
+     * 用户登录消息
+     * @param userId
+     * @return
+     */
+    public static UserActionMessage loginMessage(String userId) {
+        return new UserActionMessage(userId, UserAction.LOGIN);
+    }
 }

src/main/java/cn/meteor/beyondclouds/core/redis/RedisKey.java

@@ -12,6 +12,9 @@ public final class RedisKey {
 
     private static final String KEY_PREFIX_EMAIL_ACTIVE_CODE = "email_active_code:";
 
+    private static final String KEY_PREFIX_USER_FOLLOWED = "user_followed:";
+    private static final String KEY_PREFIX_USER_FANS = "user_fans:";
+
     /**
      * 手机验证码
      * @param mobile
@@ -27,7 +30,7 @@ public final class RedisKey {
      * @return
      */
     public static String EMAIL_VERIFY_CODE(String email) {
-        return KEY_PREFIX_MOBILE_VERIFY_CODE + email;
+        return KEY_PREFIX_EMAIL_VERIFY_CODE + email;
     }
 
     /**
@@ -38,4 +41,22 @@ public final class RedisKey {
     public static String EMAIL_ACTIVE_CODE(String activeCode) {
         return KEY_PREFIX_EMAIL_ACTIVE_CODE + activeCode;
     }
+
+    /**
+     * 用户关注的人
+     * @param userId
+     * @return
+     */
+    public static String USER_FOLLOWED(String userId) {
+        return KEY_PREFIX_USER_FOLLOWED + userId;
+    }
+
+    /**
+     * 用户的所有粉丝
+     * @param userId
+     * @return
+     */
+    public static String USER_FANS(String userId) {
+        return KEY_PREFIX_USER_FANS + userId;
+    }
 }

src/main/java/cn/meteor/beyondclouds/modules/queue/service/IMessageQueueService.java

@@ -2,6 +2,7 @@ package cn.meteor.beyondclouds.modules.queue.service;
 
 
 import cn.meteor.beyondclouds.core.queue.message.DataItemChangeMessage;
+import cn.meteor.beyondclouds.core.queue.message.UserActionMessage;
 
 /**
  * @author meteor
@@ -13,4 +14,10 @@ public interface IMessageQueueService {
      * @param itemUpdateMessage
      */
     void sendItemUpdateMessage(DataItemChangeMessage itemUpdateMessage);
+
+    /**
+     * 发送用户行为消息
+     * @param userActionMessage
+     */
+    void sendUserActionMessage(UserActionMessage userActionMessage);
 }

src/main/java/cn/meteor/beyondclouds/modules/queue/service/impl/MessageQueueServiceImpl.java

@@ -2,6 +2,7 @@ package cn.meteor.beyondclouds.modules.queue.service.impl;
 
 import cn.meteor.beyondclouds.config.properties.BeyondCloudsKafkaTopicProperties;
 import cn.meteor.beyondclouds.core.queue.message.DataItemChangeMessage;
+import cn.meteor.beyondclouds.core.queue.message.UserActionMessage;
 import cn.meteor.beyondclouds.modules.queue.service.IMessageQueueService;
 import cn.meteor.beyondclouds.util.JsonUtils;
 import lombok.extern.slf4j.Slf4j;
@@ -36,4 +37,15 @@ public class MessageQueueServiceImpl implements IMessageQueueService {
            log.error("Kafka消息发送失败：{}", e.getMessage());
        }
     }
+
+    @Override
+    public void sendUserActionMessage(UserActionMessage userActionMessage) {
+        try {
+            kafkaTemplate.send(topicProperties.getUserAction(), JsonUtils.toJson(userActionMessage));
+            log.debug("发送kafka消息：{}", userActionMessage.toString());
+        } catch (Exception e) {
+            e.printStackTrace();
+            log.error("Kafka消息发送失败：{}", e.getMessage());
+        }
+    }
 }

src/main/java/cn/meteor/beyondclouds/modules/user/listener/UserLoginAndLogoutListener.java

+package cn.meteor.beyondclouds.modules.user.listener;
+
+import cn.meteor.beyondclouds.common.helper.IRedisHelper;
+import cn.meteor.beyondclouds.core.listener.UserActionListener;
+import cn.meteor.beyondclouds.core.queue.message.UserActionMessage;
+import cn.meteor.beyondclouds.core.redis.RedisKey;
+import cn.meteor.beyondclouds.modules.user.entity.UserFollow;
+import cn.meteor.beyondclouds.modules.user.service.IUserFollowService;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 用户登录注销监听器
+ * @author meteor
+ */
+@Slf4j
+@Component
+public class UserLoginAndLogoutListener implements UserActionListener {
+
+    private IUserFollowService userFollowService;
+
+    private IRedisHelper redisHelper;
+
+    @Autowired
+    public UserLoginAndLogoutListener(IUserFollowService userFollowService, IRedisHelper redisHelper) {
+        this.userFollowService = userFollowService;
+        this.redisHelper = redisHelper;
+    }
+
+    @Override
+    public void onUserLogin(UserActionMessage userActionMessage) {
+
+        String userId = userActionMessage.getUserId();
+
+        // 查找用户关注的所有人
+        QueryWrapper<UserFollow> userFollowQueryWrapper = new QueryWrapper<>();
+        userFollowQueryWrapper.eq("follower_id", userId);
+        List<UserFollow> userFollowList = userFollowService.list(userFollowQueryWrapper);
+        List<String> followedUserIds = userFollowList.stream()
+                .map(UserFollow::getFollowedId)
+                .collect(Collectors.toList());
+
+        // 查找用户的所以粉丝
+        userFollowQueryWrapper = new QueryWrapper<>();
+        userFollowQueryWrapper.eq("followed_id", userId);
+        userFollowList = userFollowService.list(userFollowQueryWrapper);
+        List<String> fansIds = userFollowList.stream()
+                .map(UserFollow::getFollowerId)
+                .collect(Collectors.toList());
+
+
+        // 将用户的关注列表和粉丝列表存入redis
+        redisHelper.del(RedisKey.USER_FOLLOWED(userId));
+        redisHelper.del(RedisKey.USER_FANS(userId));
+
+        if (!CollectionUtils.isEmpty(followedUserIds)) {
+            redisHelper.setAdd(RedisKey.USER_FOLLOWED(userId), followedUserIds.toArray(new String[0]));
+        }
+
+        if (!CollectionUtils.isEmpty(fansIds)) {
+            redisHelper.setAdd(RedisKey.USER_FANS(userId), fansIds.toArray(new String[0]));
+        }
+        log.debug("已经将用户{}的关注列表和粉丝列表存入redis.", userId);
+    }
+
+    @Override
+    public void onUserLogout(UserActionMessage userActionMessage) {
+        String userId = userActionMessage.getUserId();
+
+        // 将用户的关注列表和粉丝列表从redis删除
+        redisHelper.del(RedisKey.USER_FOLLOWED(userId));
+        redisHelper.del(RedisKey.USER_FANS(userId));
+    }
+}

src/main/java/cn/meteor/beyondclouds/modules/user/service/impl/AuthenticationServiceImpl.java

@@ -4,7 +4,9 @@ import cn.meteor.beyondclouds.common.dto.QQAuthResultDTO;
 import cn.meteor.beyondclouds.common.exception.QQAuthenticationException;
 import cn.meteor.beyondclouds.common.helper.IQQAuthenticationHelper;
 import cn.meteor.beyondclouds.common.helper.IRedisHelper;
+import cn.meteor.beyondclouds.core.queue.message.UserActionMessage;
 import cn.meteor.beyondclouds.core.redis.RedisKey;
+import cn.meteor.beyondclouds.modules.queue.service.IMessageQueueService;
 import cn.meteor.beyondclouds.modules.user.dto.AuthenticationResultDTO;
 import cn.meteor.beyondclouds.modules.user.entity.UserAuthApp;
 import cn.meteor.beyondclouds.modules.user.entity.UserAuthLocal;
@@ -43,6 +45,8 @@ public class AuthenticationServiceImpl implements IAuthenticationService {
 
     private IRedisHelper redisHelper;
 
+    private IMessageQueueService messageQueueService;
+
     @Autowired
     public AuthenticationServiceImpl(IUserAuthLocalService userAuthLocalService,
                                      IQQAuthenticationHelper iqqAuthenticationHelper,
@@ -56,6 +60,11 @@ public class AuthenticationServiceImpl implements IAuthenticationService {
         this.redisHelper = redisHelper;
     }
 
+    @Autowired
+    public void setMessageQueueService(IMessageQueueService messageQueueService) {
+        this.messageQueueService = messageQueueService;
+    }
+
     @Override
     public AuthenticationResultDTO localAuthentication(String account, String password) throws AuthenticationServiceException {
         //1. 查找用户是否存在
@@ -84,7 +93,7 @@ public class AuthenticationServiceImpl implements IAuthenticationService {
         }
 
         // 根据userId生成token并返回
-        return makeAuthenticationResult(userAuthLocal.getUserId());
+        return makeAuthenticationResultAndSendLoginMessage(userAuthLocal.getUserId());
     }
 
     @Override
@@ -108,7 +117,7 @@ public class AuthenticationServiceImpl implements IAuthenticationService {
             }
 
             // 根据userId生成token并返回
-            return makeAuthenticationResult(userId);
+            return makeAuthenticationResultAndSendLoginMessage(userId);
 
         } catch (QQAuthenticationException e) {
             e.printStackTrace();
@@ -136,7 +145,7 @@ public class AuthenticationServiceImpl implements IAuthenticationService {
 
 
         // 根据userId生成token并返回
-        return makeAuthenticationResult(userAuthLocal.getUserId());
+        return makeAuthenticationResultAndSendLoginMessage(userAuthLocal.getUserId());
 
     }
 
@@ -145,14 +154,24 @@ public class AuthenticationServiceImpl implements IAuthenticationService {
      * @param userId
      * @return
      */
-    private AuthenticationResultDTO makeAuthenticationResult(String userId) {
+    private AuthenticationResultDTO makeAuthenticationResultAndSendLoginMessage(String userId) {
         AuthenticationResultDTO result = new AuthenticationResultDTO();
         result.setUserId(userId);
 
-        Map<String, String> claimMap = new HashMap<>();
+        Map<String, String> claimMap = new HashMap<>(1);
         claimMap.put("id", userId);
         result.setAccessToken(JwtUtils.sign(claimMap));
+
+        sendUserLoginMessage(userId);
         return result;
     }
 
+    /**
+     * 发送用户登录消息
+     * @param userId
+     */
+    private void sendUserLoginMessage(String userId) {
+        messageQueueService.sendUserActionMessage(UserActionMessage.loginMessage(userId));
+    }
+
 }