u

code/link56/link56-backend/src/main/java/cn/link56/common/config/WebMvcConfig.java

@@ -24,6 +24,8 @@ public class WebMvcConfig implements WebMvcConfigurer {
                         "/company/recommend",
                         "/company/recommend/hot",
 
+                        "/post/tag",
+
                         "/forum/post/page",
                         "/forum/post/detail",
                         "/forum/post/getRecommendList",

code/link56/link56-backend/src/main/java/cn/link56/im/NettyServer.java

@@ -1,7 +1,11 @@
 package cn.link56.im;
 
 import cn.link56.im.config.NettyConfig;
+import cn.link56.im.handler.AuthHandler;
+import cn.link56.im.handler.DispatchHandler;
+import cn.link56.im.handler.MessageHandler;
 import cn.link56.im.handler.ServerHandler;
+import io.netty.handler.timeout.IdleStateHandler;
 import org.springframework.stereotype.Component;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
@@ -18,6 +22,7 @@ import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import javax.annotation.Resource;
+import java.util.concurrent.TimeUnit;
 
 @Component
 public class NettyServer {
@@ -27,6 +32,13 @@ public class NettyServer {
     private EventLoopGroup bossGroup;
     private EventLoopGroup workerGroup;
 
+    @Resource
+    private AuthHandler authHandler;
+    @Resource
+    private ServerHandler serverHandler;
+    @Resource
+    private DispatchHandler dispatchHandler;
+
     @PostConstruct
     public void start() {
         new Thread(this::runServer, "Netty-ImServer-Thread").start();
@@ -52,7 +64,17 @@ public class NettyServer {
                             pipeline.addLast(new ChunkedWriteHandler());
                             pipeline.addLast(new CorsHandler(CorsConfig.withAnyOrigin().build()));
                             pipeline.addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath(), null, true, 65536));
-                            pipeline.addLast(new ServerHandler());
+                            // 4. 空闲检测（60 秒没收到客户端任何消息就断）
+                            pipeline.addLast("idle-handler",
+                                    new IdleStateHandler(
+                                            60,
+                                            0,
+                                            0,
+                                            TimeUnit.SECONDS
+                                    ));
+                            pipeline.addLast(serverHandler);
+                            pipeline.addLast(authHandler);
+                            pipeline.addLast(dispatchHandler);
                         }
                     });
 

code/link56/link56-backend/src/main/java/cn/link56/im/enums/MessageTypeEnums.java

@@ -2,6 +2,29 @@ package cn.link56.im.enums;
 
 import lombok.Getter;
 
+@Getter
 public enum MessageTypeEnums {
-    PING,PONG,ONLINE,OFFLINE,PRIVATE,GROUP
+    PING(0),
+    PONG(1),
+    LOGIN(2),
+    LOGOUT(3),
+    ONLINE(4),
+    OFFLINE(5),
+    PRIVATE(6),
+    GROUP(7);
+
+    private final int code;
+
+    MessageTypeEnums(int code) {
+        this.code = code;
+    }
+
+    public static MessageTypeEnums valueOf(int code) {
+        for (MessageTypeEnums type : values()) {
+            if (type.code == code) {
+                return type;
+            }
+        }
+        throw new IllegalArgumentException("未知消息类型: " + code);
+    }
 }

code/link56/link56-backend/src/main/java/cn/link56/im/handler/AuthHandler.java

@@ -0,0 +1,49 @@
+package cn.link56.im.handler;
+
+import cn.link56.auth.token.model.TokenPayload;
+import cn.link56.auth.token.service.AuthTokenService;
+import cn.link56.im.enums.MessageTypeEnums;
+import cn.link56.im.message.MessageRequest;
+import cn.link56.im.model.dto.OnlineDTO;
+import cn.link56.im.storage.SessionManager;
+import com.alibaba.druid.util.StringUtils;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@ChannelHandler.Sharable
+@Component
+public class AuthHandler extends ChannelInboundHandlerAdapter {
+    @Resource
+    private AuthTokenService authTokenService;
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg)
+            throws Exception {
+
+        if (!(msg instanceof MessageRequest)) {
+            ctx.fireChannelRead(msg);
+            return;
+        }
+
+
+        MessageRequest req = (MessageRequest) msg;
+        Integer type = req.getType();
+
+        if (type == MessageTypeEnums.PING.getCode() || type == MessageTypeEnums.ONLINE.getCode()){
+            ctx.fireChannelRead(req);
+            return;
+        }
+
+        // 如果该通道未验证
+        if (!SessionManager.isAuthenticated(ctx.channel())) {
+            ctx.close();
+            return;
+        }
+
+        ctx.fireChannelRead(req);
+    }
+}

code/link56/link56-backend/src/main/java/cn/link56/im/handler/DispatchHandler.java

@@ -0,0 +1,26 @@
+package cn.link56.im.handler;
+
+import cn.link56.im.message.MessageRequest;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 转发消息
+ */
+@ChannelHandler.Sharable
+@Component
+public class DispatchHandler extends SimpleChannelInboundHandler<MessageRequest> {
+    @Resource
+    private MessageHandler messageHandler;
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx,
+                                MessageRequest request) {
+
+        messageHandler.handleMessage(ctx.channel(), request);
+    }
+}

code/link56/link56-backend/src/main/java/cn/link56/im/handler/MessageHandler.java

@@ -1,18 +1,28 @@
 package cn.link56.im.handler;
 
+import cn.link56.auth.token.model.TokenPayload;
+import cn.link56.auth.token.service.AuthTokenService;
 import cn.link56.im.enums.MessageTypeEnums;
 import cn.link56.im.message.MessageRequest;
 import cn.link56.im.message.MessageResponse;
+import cn.link56.im.model.dto.OnlineDTO;
+import cn.link56.im.storage.SessionManager;
+import cn.link56.im.utils.MessageDataConverter;
 import cn.link56.im.utils.MessageWriter;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
+
 @Component
 public class MessageHandler {
+    @Resource
+    private AuthTokenService authTokenService;
 
     // 处理消息
     public void handleMessage(Channel channel, MessageRequest request) {
+        System.out.println(request);
         switch (MessageTypeEnums.valueOf(request.getType())) {
             case PING:
                 handlePing(channel, request);
@@ -41,22 +51,32 @@ public class MessageHandler {
 
     // 上线
     private void handleOnline(Channel channel, MessageRequest request) {
-        // 验证
-        Long from = request.getFrom();
-        Object data = request.getData(); // companyId userId
+        OnlineDTO dto = MessageDataConverter.convert(request.getData(), OnlineDTO.class);
+        TokenPayload tokenPayload = authTokenService.parsePayload(dto.getToken());
 
-        // 账号 Token
+        if (tokenPayload == null) {
+            MessageWriter.writeToChannel(channel,MessageResponse.error("Token无效"));
+            return;
+        }
+
+        Long companyId = tokenPayload.getCompanyId();
+        Long userId = tokenPayload.getUserId();
+
+        // 添加 Session 并自动标记已认证
+        SessionManager.addSession(companyId, userId, channel);
+
+        MessageWriter.writeToChannel(channel, MessageResponse.success(MessageTypeEnums.ONLINE));
     }
 
+    // 下线逻辑
     private void handleOffline(Channel channel, MessageRequest request) {
-        // 下线逻辑
     }
 
+    // 私聊逻辑
     private void handlePrivate(Channel channel, MessageRequest request) {
-        // 私聊逻辑
     }
 
+    // 群聊逻辑
     private void handleGroup(Channel channel, MessageRequest request) {
-        // 群聊逻辑
     }
 }

code/link56/link56-backend/src/main/java/cn/link56/im/handler/ServerHandler.java

@@ -1,19 +1,20 @@
 package cn.link56.im.handler;
 
 import cn.link56.im.message.MessageRequest;
+import cn.link56.im.storage.SessionManager;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.timeout.IdleStateEvent;
+import org.springframework.stereotype.Component;
 
+@Component
 @ChannelHandler.Sharable
 public class ServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
     private static final ObjectMapper MAPPER = new ObjectMapper();
 
-    private final MessageHandler messageHandler = new MessageHandler();
-
     @Override
     public void handlerAdded(ChannelHandlerContext ctx) {
         System.out.println("客户端连接: " + ctx.channel().id().asShortText());
@@ -22,13 +23,15 @@ public class ServerHandler extends SimpleChannelInboundHandler<TextWebSocketFram
     @Override
     public void handlerRemoved(ChannelHandlerContext ctx) {
         System.out.println("客户端断开: " + ctx.channel().id().asShortText());
+
+        SessionManager.removeChannel(ctx.channel());
     }
 
     @Override
-    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
-        MessageRequest request =  MAPPER.readValue(textWebSocketFrame.text(), MessageRequest.class);
+    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
+        MessageRequest request =  MAPPER.readValue(frame.text(), MessageRequest.class);
 
-        messageHandler.handleMessage(channelHandlerContext.channel(),request);
+        ctx.fireChannelRead(request);
     }
 
     @Override

code/link56/link56-backend/src/main/java/cn/link56/im/message/MessageRequest.java

@@ -3,12 +3,14 @@ package cn.link56.im.message;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
 
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
+@ToString
 public class MessageRequest {
-    private String type;
+    private Integer type;
     private Long from;
     private Object to;
     private Long timestamp;

code/link56/link56-backend/src/main/java/cn/link56/im/message/MessageResponse.java

@@ -13,7 +13,7 @@ import lombok.NoArgsConstructor;
 @NoArgsConstructor
 @Builder
 public class MessageResponse {
-    private String type;
+    private Integer type;
     private Long from;
     private Object data;
     private Long timestamp;
@@ -27,7 +27,7 @@ public class MessageResponse {
     }
 
     public static MessageResponse success(MessageTypeEnums messageTypeEnums) {
-        return buildResponse(messageTypeEnums.name(), MessageResponseStatusConstants.SUCCESS, null, null);
+        return buildResponse(messageTypeEnums.getCode(), MessageResponseStatusConstants.SUCCESS, null, null);
     }
 
     public static MessageResponse error() {
@@ -38,7 +38,7 @@ public class MessageResponse {
         return buildResponse(null, MessageResponseStatusConstants.ERROR, null, msg);
     }
 
-    private static MessageResponse buildResponse(String type, Integer code, Object data, String message) {
+    private static MessageResponse buildResponse(Integer type, Integer code, Object data, String message) {
         MessageResponse response = new MessageResponse();
         response.setType(type);
         response.setCode(code);

code/link56/link56-backend/src/main/java/cn/link56/im/model/dto/OnlineDTO.java

@@ -0,0 +1,12 @@
+package cn.link56.im.model.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class OnlineDTO {
+    private String token;
+}

code/link56/link56-backend/src/main/java/cn/link56/im/storage/SessionManager.java

@@ -1,41 +1,87 @@
 package cn.link56.im.storage;
 
 import io.netty.channel.Channel;
+import io.netty.util.AttributeKey;
 
 import java.util.concurrent.ConcurrentHashMap;
 
 public class SessionManager {
+
+    /** userId -> Session */
     private static final ConcurrentHashMap<Long, Session> sessions = new ConcurrentHashMap<>();
 
+    /** Channel 是否已认证 */
+    private static final AttributeKey<Boolean> AUTH_KEY =
+            AttributeKey.valueOf("AUTHENTICATED");
+
+    /** Channel 绑定的 userId */
+    private static final AttributeKey<Long> USER_ID_KEY =
+            AttributeKey.valueOf("USER_ID");
+
+    /** Channel 绑定的 companyId */
+    private static final AttributeKey<Long> COMPANY_ID_KEY =
+            AttributeKey.valueOf("COMPANY_ID");
+
+    /* ================= 认证相关 ================= */
+
+    public static void markAuthenticated(Channel channel, Long companyId, Long userId) {
+        if (channel == null || companyId == null || userId == null) return;
+
+        channel.attr(AUTH_KEY).set(true);
+        channel.attr(USER_ID_KEY).set(userId);
+        channel.attr(COMPANY_ID_KEY).set(companyId);
+    }
+
+    public static boolean isAuthenticated(Channel channel) {
+        if (channel == null) return false;
+        return Boolean.TRUE.equals(channel.attr(AUTH_KEY).get());
+    }
+
+    public static Long getUserId(Channel channel) {
+        return channel.attr(USER_ID_KEY).get();
+    }
+
+    public static Long getCompanyId(Channel channel) {
+        return channel.attr(COMPANY_ID_KEY).get();
+    }
+
+    /* ================= Session 管理 ================= */
 
-    /** 添加会话，支持同一个用户多端登录 */
     public static void addSession(Long companyId, Long userId, Channel channel) {
         if (companyId == null || userId == null || channel == null) return;
 
-        // 通过 compute 方法保证线程安全
+        // 线程安全地添加 Channel 到 Session
         sessions.compute(userId, (key, session) -> {
             if (session == null) {
-                // 新建 Session
                 session = new Session(companyId, userId);
             }
-            // 添加新的 Channel 到 Session
             session.addChannel(channel);
+
+            markAuthenticated(channel, companyId, userId);
+
             return session;
         });
     }
 
-    /** 移除某个 Channel */
-    public static void removeChannel(Long userId, Channel channel) {
-        if (userId == null || channel == null) return;
+
+    public static void removeChannel(Channel channel) {
+        if (channel == null) return;
+
+        Long userId = getUserId(channel);
+        if (userId == null) return;
 
         Session session = sessions.get(userId);
         if (session != null) {
             session.removeChannel(channel);
-            // 如果用户所有连接都断开，删除 Session
             if (!session.isOnline()) {
                 sessions.remove(userId);
             }
         }
+
+        // 清理 Channel 上的状态
+        channel.attr(AUTH_KEY).set(null);
+        channel.attr(USER_ID_KEY).set(null);
+        channel.attr(COMPANY_ID_KEY).set(null);
     }
 
     /** 获取用户 Session */
@@ -43,7 +89,6 @@ public class SessionManager {
         return sessions.get(userId);
     }
 
-    /** 检查用户是否在线 */
     public static boolean isOnline(Long userId) {
         Session session = sessions.get(userId);
         return session != null && session.isOnline();

code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageDataConverter.java

@@ -0,0 +1,11 @@
+package cn.link56.im.utils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class MessageDataConverter {
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    public static <T> T convert(Object data, Class<T> clazz) {
+        return MAPPER.convertValue(data, clazz);
+    }
+}

code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageWriter.java

@@ -14,18 +14,20 @@ import java.util.Set;
 public class MessageWriter {
     private static final ObjectMapper MAPPER = new ObjectMapper();
 
-
-    public static void writeToUser(Long userId,Object message){
-        if (ObjectUtils.isEmpty(userId) || ObjectUtils.isEmpty(message)){
+    public static void writeToUser(Long userId, Object message) {
+        if (userId == null || message == null) {
             return;
         }
 
         Session session = SessionManager.getSession(userId);
-        Set<Channel> channels = session.getChannels();
-        if (channels.isEmpty()) return;
+        if (session == null || session.getChannels().isEmpty()) {
+            return;
+        }
 
-        for (Channel ch : channels) {
-            writeToChannel(ch, message);
+        for (Channel ch : session.getChannels()) {
+            if (ch != null && ch.isActive()) {
+                writeToChannel(ch, message);
+            }
         }
     }
 
@@ -40,6 +42,8 @@ public class MessageWriter {
             String payload = MAPPER.writeValueAsString(message);
             // 发送给客户端
             channel.writeAndFlush(new TextWebSocketFrame(payload));
+            System.out.println("===发送消息===");
+            System.out.println(payload);
         } catch (Exception e) {
             System.out.println("===发送消息失败===");
             e.printStackTrace();

code/link56/link56-backend/src/main/resources/application.yml

@@ -35,7 +35,7 @@ mybatis-plus:
     cache-enabled: true
     use-generated-keys: true
     default-executor-type: simple
-    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+#    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
     map-underscore-to-camel-case: true
   global-config:
     db-config: