From 5c0ec34760373aaa8a505251d53b06da8d7dbd8f Mon Sep 17 00:00:00 2001 From: zhongzb <972627721@qq.com> Date: Sun, 20 Aug 2023 14:37:01 +0800 Subject: [PATCH] =?UTF-8?q?fix:=201.=E7=99=BB=E5=BD=95=E9=9B=86=E7=BE=A4?= =?UTF-8?q?=E6=94=B9=E9=80=A0=202.netty=E5=8F=96=E6=B6=88=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/mallchat.sql | 155 ++++++++++++++++-- .../domain/vo/response/ChatMessageResp.java | 2 + .../common/common/constant/MQConstant.java | 8 +- .../common/domain/dto/LoginMessageDTO.java | 17 +- .../domain/dto/ScanSuccessMessageDTO.java | 12 +- .../common/common/utils/CacheHolder.java | 26 --- .../user/consumer/MsgLoginConsumer.java | 33 +--- .../user/consumer/ScanSuccessConsumer.java | 24 +-- .../user/controller/WxPortalController.java | 3 +- .../custom/user/service/WebSocketService.java | 9 +- .../custom/user/service/WxMsgService.java | 57 +------ .../service/impl/WebSocketServiceImpl.java | 91 ++++------ .../user/websocket/NettyCollectorHandler.java | 27 --- .../user/websocket/NettyWebSocketServer.java | 1 - .../NettyWebSocketServerHandler.java | 4 - 15 files changed, 197 insertions(+), 272 deletions(-) delete mode 100644 mallchat-common/src/main/java/com/abin/mallchat/common/common/utils/CacheHolder.java delete mode 100644 mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyCollectorHandler.java diff --git a/docs/mallchat.sql b/docs/mallchat.sql index f2db654..6bea229 100644 --- a/docs/mallchat.sql +++ b/docs/mallchat.sql @@ -95,24 +95,26 @@ INSERT INTO `room` VALUES (1, '抹茶群聊', 1, '2023-03-25 22:30:07.328', '202 -- Table structure for user -- ---------------------------- DROP TABLE IF EXISTS `user`; -CREATE TABLE `user` ( - `id` bigint(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '用户id', - `name` varchar(20) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '用户昵称', - `avatar` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '用户头像', - `sex` int(11) NULL DEFAULT NULL COMMENT '性别 1为男性，2为女性', - `open_id` char(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '微信openid用户标识', +CREATE TABLE `user` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '用户id', + `name` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '用户昵称', + `avatar` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '用户头像', + `sex` int(11) DEFAULT NULL COMMENT '性别 1为男性，2为女性', + `open_id` char(32) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '微信openid用户标识', + `active_status` int(11) DEFAULT '2' COMMENT '在线状态 1在线 2离线', `last_opt_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '最后上下线时间', - `ip_info` json NULL COMMENT 'ip信息', - `item_id` bigint(20) NULL DEFAULT NULL COMMENT '佩戴的徽章id', - `status` int(11) DEFAULT "0" COMMENT '使用状态 0.正常 1拉黑', + `ip_info` json DEFAULT NULL COMMENT 'ip信息', + `item_id` bigint(20) DEFAULT NULL COMMENT '佩戴的徽章id', + `status` int(11) DEFAULT '0' COMMENT '使用状态 0.正常 1拉黑', `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', PRIMARY KEY (`id`) USING BTREE, - UNIQUE INDEX `uniq_open_id`(`open_id`) USING BTREE, - UNIQUE INDEX `uniq_name`(`name`) USING BTREE, - INDEX `idx_create_time`(`create_time`) USING BTREE, - INDEX `idx_update_time`(`update_time`) USING BTREE -) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '用户表' ROW_FORMAT = Dynamic; + UNIQUE KEY `uniq_open_id` (`open_id`) USING BTREE, + UNIQUE KEY `uniq_name` (`name`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE, + KEY `idx_active_status_last_opt_time` (`active_status`,`last_opt_time`) +) ENGINE=InnoDB AUTO_INCREMENT=20000 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='用户表'; INSERT INTO `mallchat`.`user` (`id`, `name`, `avatar`, `sex`, `open_id`, `last_opt_time`, `ip_info`, `item_id`, `status`, `create_time`, `update_time`) VALUES (10001, 'ChatGPT', 'https://img1.baidu.com/it/u=3613958228,3522035000&fm=253&fmt=auto&app=120&f=JPEG?w=500&h=500', 0, '??', '2023-06-29 17:03:03.357', NULL, NULL, 0, '2023-06-29 17:03:03.357', '2023-07-01 14:56:10.271'); INSERT INTO `mallchat`.`user` (`id`, `name`, `avatar`, `sex`, `open_id`, `last_opt_time`, `ip_info`, `item_id`, `status`, `create_time`, `update_time`) VALUES (10002, 'ChatGLM2', 'http://mms1.baidu.com/it/u=1979830414,2984779047&fm=253&app=138&f=JPEG&fmt=auto&q=75?w=500&h=500', NULL, '450', '2023-07-01 11:58:24.605', NULL, NULL, 0, '2023-07-01 11:58:24.605', '2023-07-01 12:02:56.900'); -- ---------------------------- @@ -202,4 +204,127 @@ CREATE TABLE `user_emoji` ( `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', PRIMARY KEY (`id`) USING BTREE, KEY `IDX_USER_EMOJIS_UID` (`uid`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='用户表情包'; \ No newline at end of file +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='用户表情包'; +###单聊群聊功能 +DROP TABLE IF EXISTS `user_apply`; +CREATE TABLE `user_apply` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `uid` bigint(20) NOT NULL COMMENT '申请人uid', + `type` int(11) NOT NULL COMMENT '申请类型 1加好友', + `target_id` bigint(20) NOT NULL COMMENT '接收人uid', + `msg` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '申请信息', + `status` int(11) NOT NULL COMMENT '申请状态 1待审批 2同意', + `read_status` int(11) NOT NULL COMMENT '阅读状态 1未读 2已读', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + KEY `idx_target_id_uid_status` (`target_id`,`uid`,`status`) USING BTREE, + KEY `idx_target_id` (`target_id`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户申请表'; +DROP TABLE IF EXISTS `user_friend`; +CREATE TABLE `user_friend` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `uid` bigint(20) NOT NULL COMMENT 'uid', + `friend_uid` bigint(20) NOT NULL COMMENT '好友uid', + `delete_status` int(1) NOT NULL DEFAULT '0' COMMENT '逻辑删除(0-正常,1-删除)', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + KEY `idx_uid_friend_uid` (`uid`,`friend_uid`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户联系人表'; +DROP TABLE IF EXISTS `room`; +CREATE TABLE `room` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `type` int(11) NOT NULL COMMENT '房间类型 1群聊 2单聊', + `hot_flag` int(11) DEFAULT '0' COMMENT '是否全员展示 0否 1是', + `active_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '群最后消息的更新时间（热点群不需要写扩散，只更新这里）', + `last_msg_id` bigint(20) DEFAULT NULL COMMENT '会话中的最后一条消息id', + `ext_json` json DEFAULT NULL COMMENT '额外信息（根据不同类型房间有不同存储的东西）', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='房间表'; +DROP TABLE IF EXISTS `room_friend`; +CREATE TABLE `room_friend` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `room_id` bigint(20) NOT NULL COMMENT '房间id', + `uid1` bigint(20) NOT NULL COMMENT 'uid1（更小的uid）', + `uid2` bigint(20) NOT NULL COMMENT 'uid2（更大的uid）', + `room_key` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '房间key由两个uid拼接，先做排序uid1_uid2', + `status` int(11) NOT NULL COMMENT '房间状态 0正常 1禁用(删好友了禁用)', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `room_key` (`room_key`) USING BTREE, + KEY `idx_room_id` (`room_id`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='单聊房间表'; +DROP TABLE IF EXISTS `room_group`; +CREATE TABLE `room_group` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `room_id` bigint(20) NOT NULL COMMENT '房间id', + `name` varchar(16) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '群名称', + `avatar` varchar(256) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '群头像', + `ext_json` json DEFAULT NULL COMMENT '额外信息（根据不同类型房间有不同存储的东西）', + `delete_status` int(1) NOT NULL DEFAULT '0' COMMENT '逻辑删除(0-正常,1-删除)', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + KEY `idx_room_id` (`room_id`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='群聊房间表'; +DROP TABLE IF EXISTS `group_member`; +CREATE TABLE `group_member` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `group_id` bigint(20) NOT NULL COMMENT '群主id', + `uid` bigint(20) NOT NULL COMMENT '成员uid', + `role` int(11) NOT NULL COMMENT '成员角色 1群主 2管理员 3普通成员', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + KEY `idx_group_id_role` (`group_id`,`role`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='群成员表'; +DROP TABLE IF EXISTS `contact`; +CREATE TABLE `contact` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `uid` bigint(20) NOT NULL COMMENT 'uid', + `room_id` bigint(20) NOT NULL COMMENT '房间id', + `read_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '阅读到的时间', + `active_time` datetime(3) DEFAULT NULL COMMENT '会话内消息最后更新的时间(只有普通会话需要维护，全员会话不需要维护)', + `last_msg_id` bigint(20) DEFAULT NULL COMMENT '会话最新消息id', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `uniq_uid_room_id` (`uid`,`room_id`) USING BTREE, + KEY `idx_room_id_read_time` (`room_id`,`read_time`) USING BTREE, + KEY `idx_create_time` (`create_time`) USING BTREE, + KEY `idx_update_time` (`update_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='会话列表'; + +INSERT INTO `user` (`id`, `name`, `avatar`, `sex`, `open_id`, `last_opt_time`, `ip_info`, `item_id`, `status`, `create_time`, `update_time`) VALUES (1, '系统消息', 'http://mms1.baidu.com/it/u=1979830414,2984779047&fm=253&app=138&f=JPEG&fmt=auto&q=75?w=500&h=500', NULL, '0', '2023-07-01 11:58:24.605', NULL, NULL, 0, '2023-07-01 11:58:24.605', '2023-07-01 12:02:56.900'); +insert INTO `room`(`id`,`type`,`hot_flag`) values (1,1,1); +insert INTO `room_group`(`id`,`room_id`,`name`,`avatar`) values (1,1,'抹茶全员群','https://mallchat.cn/assets/logo-e81cd252.jpeg'); +DROP TABLE IF EXISTS `secure_invoke_record`; +CREATE TABLE `secure_invoke_record` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id', + `secure_invoke_json` json NOT NULL COMMENT '请求快照参数json', + `status` tinyint(8) NOT NULL COMMENT '状态 1待执行 2已失败', + `next_retry_time` datetime(3) NOT NULL COMMENT '下一次重试的时间', + `retry_times` int(11) NOT NULL COMMENT '已经重试的次数', + `max_retry_times` int(11) NOT NULL COMMENT '最大重试次数', + `fail_reason` text COMMENT '执行失败的堆栈', + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间', + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '修改时间', + PRIMARY KEY (`id`) USING BTREE, + KEY `idx_next_retry_time` (`next_retry_time`) USING BTREE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='本地消息表'; \ No newline at end of file diff --git a/mallchat-common/src/main/java/com/abin/mallchat/common/chat/domain/vo/response/ChatMessageResp.java b/mallchat-common/src/main/java/com/abin/mallchat/common/chat/domain/vo/response/ChatMessageResp.java index 87758ca..5dd3306 100644 --- a/mallchat-common/src/main/java/com/abin/mallchat/common/chat/domain/vo/response/ChatMessageResp.java +++ b/mallchat-common/src/main/java/com/abin/mallchat/common/chat/domain/vo/response/ChatMessageResp.java @@ -34,6 +34,8 @@ public class ChatMessageResp { public static class Message { @ApiModelProperty("消息id") private Long id; + @ApiModelProperty("房间id") + private Long roomId; @ApiModelProperty("消息发送时间") private Date sendTime; @ApiModelProperty("消息类型 1正常文本 2.撤回消息") diff --git a/mallchat-common/src/main/java/com/abin/mallchat/common/common/constant/MQConstant.java b/mallchat-common/src/main/java/com/abin/mallchat/common/common/constant/MQConstant.java index aa877eb..a33e37d 100644 --- a/mallchat-common/src/main/java/com/abin/mallchat/common/common/constant/MQConstant.java +++ b/mallchat-common/src/main/java/com/abin/mallchat/common/common/constant/MQConstant.java @@ -20,12 +20,12 @@ public interface MQConstant { /** * (授权完成后)登录信息mq */ - String LOGIN_MSG_TOPIC = "login_send_msg"; - String LOGIN_MSG_GROUP = "login_send_msg_group"; + String LOGIN_MSG_TOPIC = "user_login_send_msg"; + String LOGIN_MSG_GROUP = "user_login_send_msg_group"; /** * 扫码成功 信息发送mq */ - String SCAN_MSG_TOPIC = "scan_send_msg"; - String SCAN_MSG_GROUP = "scan_send_msg_group"; + String SCAN_MSG_TOPIC = "user_scan_send_msg"; + String SCAN_MSG_GROUP = "user_scan_send_msg_group"; } diff --git a/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/LoginMessageDTO.java b/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/LoginMessageDTO.java index f250c0a..805d2a3 100644 --- a/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/LoginMessageDTO.java +++ b/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/LoginMessageDTO.java @@ -1,11 +1,7 @@ package com.abin.mallchat.common.common.domain.dto; -import com.abin.mallchat.common.user.domain.enums.WSBaseResp; -import com.abin.mallchat.common.user.domain.enums.WSPushTypeEnum; import lombok.AllArgsConstructor; import lombok.Data; -import lombok.NoArgsConstructor; -import me.chanjar.weixin.mp.bean.message.WxMpXmlMessage; import java.io.Serializable; @@ -15,18 +11,11 @@ import java.io.Serializable; * Date: 2023-08-12 */ @Data -@NoArgsConstructor +@AllArgsConstructor public class LoginMessageDTO implements Serializable { private static final long serialVersionUID = 1L; - /** - * 微信公众号获得扫码事件后,发送给我方的回调信息 - */ - private WxMpXmlMessage wxMpXmlMessage ; - - public LoginMessageDTO(WxMpXmlMessage wxMpXmlMessage) { - this.wxMpXmlMessage = wxMpXmlMessage; - } - + private Long uid; + private Integer code; } diff --git a/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/ScanSuccessMessageDTO.java b/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/ScanSuccessMessageDTO.java index 7472181..5ecb230 100644 --- a/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/ScanSuccessMessageDTO.java +++ b/mallchat-common/src/main/java/com/abin/mallchat/common/common/domain/dto/ScanSuccessMessageDTO.java @@ -1,7 +1,5 @@ package com.abin.mallchat.common.common.domain.dto; -import com.abin.mallchat.common.user.domain.enums.WSBaseResp; -import com.abin.mallchat.common.user.domain.enums.WSPushTypeEnum; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -17,17 +15,9 @@ import java.io.Serializable; @AllArgsConstructor @NoArgsConstructor public class ScanSuccessMessageDTO implements Serializable { - /** - * 推送的ws消息 - */ - private WSBaseResp wsBaseMsg; /** * 推送的uid */ - private Integer loginCode; + private Integer code; - public ScanSuccessMessageDTO(Integer loginCode, WSBaseResp wsBaseMsg) { - this.loginCode = loginCode; - this.wsBaseMsg = wsBaseMsg; - } } diff --git a/mallchat-common/src/main/java/com/abin/mallchat/common/common/utils/CacheHolder.java b/mallchat-common/src/main/java/com/abin/mallchat/common/common/utils/CacheHolder.java deleted file mode 100644 index 8fc4f0c..0000000 --- a/mallchat-common/src/main/java/com/abin/mallchat/common/common/utils/CacheHolder.java +++ /dev/null @@ -1,26 +0,0 @@ -package com.abin.mallchat.common.common.utils; - -import com.github.benmanes.caffeine.cache.Cache; -import com.github.benmanes.caffeine.cache.Caffeine; -import io.netty.channel.Channel; - -import java.time.Duration; - -/** - * Description: Cache管理器 - * Author: abin - * Date: 2023-04-05 - */ -public class CacheHolder { - - private static final Long MAX_MUM_SIZE = 10000L; - - private static final Duration EXPIRE_TIME = Duration.ofHours(1); - /** - * 所有请求登录的code与channel关系 - */ - public static final Cache WAIT_LOGIN_MAP = Caffeine.newBuilder() - .expireAfterWrite(EXPIRE_TIME) - .maximumSize(MAX_MUM_SIZE) - .build(); -} diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/MsgLoginConsumer.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/MsgLoginConsumer.java index 43b88e6..501b84c 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/MsgLoginConsumer.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/MsgLoginConsumer.java @@ -1,23 +1,14 @@ package com.abin.mallchat.custom.user.consumer; import com.abin.mallchat.common.common.constant.MQConstant; -import com.abin.mallchat.common.common.constant.RedisKey; import com.abin.mallchat.common.common.domain.dto.LoginMessageDTO; -import com.abin.mallchat.common.common.utils.CacheHolder; -import com.abin.mallchat.common.user.dao.UserDao; -import com.abin.mallchat.common.user.domain.entity.User; import com.abin.mallchat.custom.user.service.WebSocketService; -import com.abin.mallchat.custom.user.service.WxMsgService; -import io.netty.channel.Channel; -import me.chanjar.weixin.mp.bean.message.WxMpXmlMessage; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import java.util.Objects; - /** * Description: 在本地服务上找寻对应channel，将对应用户登陆，并触发所有用户收到上线事件 * Author: abin @@ -27,30 +18,12 @@ import java.util.Objects; @Component public class MsgLoginConsumer implements RocketMQListener { @Autowired - private WxMsgService wxMsgService; - @Autowired - private UserDao userDao; + private WebSocketService webSocketService; @Override public void onMessage(LoginMessageDTO loginMessageDTO) { - WxMpXmlMessage wxMpXmlMessage = loginMessageDTO.getWxMpXmlMessage(); - //给二维码绑定的登录code - Integer eventKey = Integer.parseInt(getEventKey(wxMpXmlMessage)); - //本地未储存对应的channel,则结束 - Channel channel = CacheHolder.WAIT_LOGIN_MAP.getIfPresent(eventKey); - if (Objects.isNull(channel)) { - return; - } - //查询openid对应的用户(必然存在) - String openid = wxMpXmlMessage.getFromUser(); - User user = userDao.getByOpenId(openid); - //登录,并且清除缓存 - wxMsgService.login(user.getId(), eventKey); - } - - private String getEventKey(WxMpXmlMessage wxMpXmlMessage) { - //扫码关注的渠道事件有前缀，需要去除 - return wxMpXmlMessage.getEventKey().replace("qrscene_", ""); + //尝试登录登录 + webSocketService.scanLoginSuccess(loginMessageDTO.getCode(), loginMessageDTO.getUid()); } } diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/ScanSuccessConsumer.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/ScanSuccessConsumer.java index 3d22798..d5f0091 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/ScanSuccessConsumer.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/consumer/ScanSuccessConsumer.java @@ -1,24 +1,14 @@ package com.abin.mallchat.custom.user.consumer; -import cn.hutool.json.JSONUtil; import com.abin.mallchat.common.common.constant.MQConstant; import com.abin.mallchat.common.common.domain.dto.ScanSuccessMessageDTO; -import com.abin.mallchat.common.common.utils.CacheHolder; -import com.abin.mallchat.common.user.dao.UserDao; -import com.abin.mallchat.common.user.domain.entity.User; import com.abin.mallchat.custom.user.service.WebSocketService; -import com.abin.mallchat.custom.user.service.WxMsgService; -import io.netty.channel.Channel; -import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; -import me.chanjar.weixin.mp.bean.message.WxMpXmlMessage; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import java.util.Objects; - /** * Description: 将扫码成功的信息发送给对应的用户,等待授权 * Author: abin @@ -29,22 +19,10 @@ import java.util.Objects; public class ScanSuccessConsumer implements RocketMQListener { @Autowired private WebSocketService webSocketService; - @Autowired - private WxMsgService wxMsgService; - @Autowired - private UserDao userDao; @Override public void onMessage(ScanSuccessMessageDTO scanSuccessMessageDTO) { - Integer loginCode = scanSuccessMessageDTO.getLoginCode(); - //本地未储存对应的channel,则结束 - Channel channel = CacheHolder.WAIT_LOGIN_MAP.getIfPresent(loginCode); - if (Objects.isNull(channel)) { - return; - } - //给正在等待登陆的channel发送扫码成功的消息，等待授权 - channel.writeAndFlush(new TextWebSocketFrame(JSONUtil.toJsonStr(scanSuccessMessageDTO.getWsBaseMsg()))); + webSocketService.scanSuccess(scanSuccessMessageDTO.getCode()); } - } diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/controller/WxPortalController.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/controller/WxPortalController.java index 84c4135..cc2cbc4 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/controller/WxPortalController.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/controller/WxPortalController.java @@ -5,7 +5,6 @@ import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import me.chanjar.weixin.common.bean.WxOAuth2UserInfo; import me.chanjar.weixin.common.bean.oauth2.WxOAuth2AccessToken; -import me.chanjar.weixin.common.error.WxErrorException; import me.chanjar.weixin.mp.api.WxMpMessageRouter; import me.chanjar.weixin.mp.api.WxMpService; import me.chanjar.weixin.mp.bean.message.WxMpXmlMessage; @@ -50,7 +49,7 @@ public class WxPortalController { } @GetMapping("/callBack") - public RedirectView callBack(@RequestParam String code) throws WxErrorException { + public RedirectView callBack(@RequestParam String code) { try { WxOAuth2AccessToken accessToken = wxService.getOAuth2Service().getAccessToken(code); WxOAuth2UserInfo userInfo = wxService.getOAuth2Service().getUserInfo(accessToken, "zh_CN"); diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WebSocketService.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WebSocketService.java index 6c93815..d849e4e 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WebSocketService.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WebSocketService.java @@ -1,6 +1,5 @@ package com.abin.mallchat.custom.user.service; -import com.abin.mallchat.common.user.domain.entity.User; import com.abin.mallchat.common.user.domain.enums.WSBaseResp; import com.abin.mallchat.common.user.domain.vo.request.ws.WSAuthorize; import io.netty.channel.Channel; @@ -37,19 +36,15 @@ public interface WebSocketService { /** * 扫码用户登录成功通知,清除本地Cache中的loginCode和channel的关系 - * - * @param loginCode - * @param user - * @param token */ - Boolean scanLoginSuccess(Integer loginCode, User user, String token); + Boolean scanLoginSuccess(Integer loginCode, Long uid); /** * 通知用户扫码成功 * * @param loginCode */ - Boolean scanSuccess(Integer loginCode, Long uid); + Boolean scanSuccess(Integer loginCode); /** * 推动消息给所有在线的人 diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WxMsgService.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WxMsgService.java index 2557068..09ff8f0 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WxMsgService.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/WxMsgService.java @@ -4,14 +4,13 @@ import cn.hutool.core.util.RandomUtil; import com.abin.mallchat.common.common.constant.MQConstant; import com.abin.mallchat.common.common.constant.RedisKey; import com.abin.mallchat.common.common.domain.dto.LoginMessageDTO; -import com.abin.mallchat.common.common.utils.CacheHolder; +import com.abin.mallchat.common.common.domain.dto.ScanSuccessMessageDTO; import com.abin.mallchat.common.common.utils.RedisUtils; import com.abin.mallchat.common.user.dao.UserDao; import com.abin.mallchat.common.user.domain.entity.User; import com.abin.mallchat.custom.user.service.adapter.TextBuilder; import com.abin.mallchat.custom.user.service.adapter.UserAdapter; import com.abin.mallchat.transaction.service.MQProducer; -import io.netty.channel.Channel; import lombok.extern.slf4j.Slf4j; import me.chanjar.weixin.common.bean.WxOAuth2UserInfo; import me.chanjar.weixin.mp.api.WxMpService; @@ -20,14 +19,11 @@ import me.chanjar.weixin.mp.bean.message.WxMpXmlOutMessage; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Lazy; import org.springframework.dao.DuplicateKeyException; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import java.net.URLEncoder; import java.util.Objects; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; /** @@ -41,22 +37,14 @@ public class WxMsgService { /** * 用户的openId和前端登录场景code的映射关系 */ - private static final ConcurrentHashMap OPENID_EVENT_CODE_MAP = new ConcurrentHashMap<>(); private static final String URL = "https://open.weixin.qq.com/connect/oauth2/authorize?appid=%s&redirect_uri=%s&response_type=code&scope=snsapi_userinfo&state=STATE#wechat_redirect"; @Value("${wx.mp.callback}") private String callback; @Autowired private UserDao userDao; @Autowired - @Lazy - private WebSocketService webSocketService; - @Autowired - private LoginService loginService; - @Autowired private UserService userService; @Autowired - private ThreadPoolTaskExecutor threadPoolTaskExecutor; - @Autowired private MQProducer mqProducer; public WxMpXmlOutMessage scan(WxMpService wxMpService, WxMpXmlMessage wxMpXmlMessage) { @@ -65,14 +53,7 @@ public class WxMsgService { User user = userDao.getByOpenId(openid); //如果已经注册,直接登录成功 if (Objects.nonNull(user) && StringUtils.isNotEmpty(user.getAvatar())) { - Channel channel = CacheHolder.WAIT_LOGIN_MAP.getIfPresent(loginCode); - //要么在本地登录,否则利用mq广播到到所有服务上尝试登录 - if (Objects.nonNull(channel)) { - String token = loginService.login(user.getId()); - webSocketService.scanLoginSuccess(loginCode, user, token); - }else { - mqProducer.sendMsg(MQConstant.LOGIN_MSG_TOPIC, new LoginMessageDTO(wxMpXmlMessage)); - } + mqProducer.sendMsg(MQConstant.LOGIN_MSG_TOPIC, new LoginMessageDTO(user.getId(), loginCode)); return null; } @@ -81,12 +62,10 @@ public class WxMsgService { user = User.builder().openId(openid).build(); userService.register(user); } - Long uid = user.getId(); - - //在 redis中保存openid和场景code的关系，后续才能通知到前端,旧版数据没有清除,这里设置了过期时间 + //在redis中保存openid和场景code的关系，后续才能通知到前端,旧版数据没有清除,这里设置了过期时间 RedisUtils.set(RedisKey.getKey(RedisKey.OPEN_ID_STRING, openid), loginCode, 60, TimeUnit.MINUTES); - //授权流程,给用户发送授权消息，并且异步通知前端扫码成功(如非本地channel,使用MQ通知某服务对前端进行通知扫码成功) - threadPoolTaskExecutor.execute(() -> webSocketService.scanSuccess(loginCode, uid)); + //授权流程,给用户发送授权消息，并且异步通知前端扫码成功,等待授权 + mqProducer.sendMsg(MQConstant.SCAN_MSG_TOPIC, new ScanSuccessMessageDTO(loginCode)); String skipUrl = String.format(URL, wxMpService.getWxMpConfigStorage().getAppId(), URLEncoder.encode(callback + "/wx/portal/public/callBack")); WxMpXmlOutMessage.TEXT().build(); return new TextBuilder().build("请点击链接授权：登录", wxMpXmlMessage, wxMpService); @@ -108,20 +87,10 @@ public class WxMsgService { if (StringUtils.isEmpty(user.getName())) { fillUserInfo(user.getId(), userInfo); } - //找到对应的 - Integer eventKey = RedisUtils.get(RedisKey.getKey(RedisKey.OPEN_ID_STRING, userInfo.getOpenid()), Integer.class); - //如果channel就在本地，直接登录 - Channel channel = CacheHolder.WAIT_LOGIN_MAP.getIfPresent(eventKey); - if (Objects.nonNull(channel)) { - login(user.getId(), eventKey); - }else { - //如果channel不在本地，利用mq广播到到所有服务上,尝试进行登录 - //手动生成一个WxMpXmlMessage - WxMpXmlMessage wxMpXmlMessage = new WxMpXmlMessage(); - wxMpXmlMessage.setFromUser(userInfo.getOpenid()); - wxMpXmlMessage.setEventKey("qrscene_"+eventKey); - mqProducer.sendMsg(MQConstant.LOGIN_MSG_TOPIC, new LoginMessageDTO(wxMpXmlMessage)); - } + //找到对应的code + Integer code = RedisUtils.get(RedisKey.getKey(RedisKey.OPEN_ID_STRING, userInfo.getOpenid()), Integer.class); + //发送登录成功事件 + mqProducer.sendMsg(MQConstant.LOGIN_MSG_TOPIC, new LoginMessageDTO(user.getId(), code)); } private void fillUserInfo(Long uid, WxOAuth2UserInfo userInfo) { @@ -138,12 +107,4 @@ public class WxMsgService { update.setName("名字重置" + RandomUtil.randomInt(100000)); } } - - public void login(Long uid, Integer eventKey) { - User user = userDao.getById(uid); - //调用用户登录模块 - String token = loginService.login(uid); - //推送前端登录成功 - webSocketService.scanLoginSuccess(eventKey, user, token); - } } diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/impl/WebSocketServiceImpl.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/impl/WebSocketServiceImpl.java index 2120b78..2d9743c 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/impl/WebSocketServiceImpl.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/service/impl/WebSocketServiceImpl.java @@ -5,12 +5,9 @@ import cn.hutool.core.util.ObjectUtil; import cn.hutool.json.JSONUtil; import com.abin.mallchat.common.common.annotation.FrequencyControl; import com.abin.mallchat.common.common.config.ThreadPoolConfig; -import com.abin.mallchat.common.common.constant.MQConstant; import com.abin.mallchat.common.common.constant.RedisKey; -import com.abin.mallchat.common.common.domain.dto.ScanSuccessMessageDTO; import com.abin.mallchat.common.common.event.UserOfflineEvent; import com.abin.mallchat.common.common.event.UserOnlineEvent; -import com.abin.mallchat.common.common.utils.CacheHolder; import com.abin.mallchat.common.common.utils.RedisUtils; import com.abin.mallchat.common.user.dao.UserDao; import com.abin.mallchat.common.user.domain.entity.User; @@ -25,6 +22,8 @@ import com.abin.mallchat.custom.user.service.WebSocketService; import com.abin.mallchat.custom.user.service.adapter.WSAdapter; import com.abin.mallchat.custom.user.websocket.NettyUtil; import com.abin.mallchat.transaction.service.MQProducer; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import io.netty.channel.Channel; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import lombok.SneakyThrows; @@ -38,12 +37,12 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import java.time.Duration; -import java.util.*; +import java.util.Date; +import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.ReentrantLock; /** * Description: websocket处理类 @@ -55,7 +54,14 @@ import java.util.concurrent.locks.ReentrantLock; public class WebSocketServiceImpl implements WebSocketService { private static final Duration EXPIRE_TIME = Duration.ofHours(1); - + private static final Long MAX_MUM_SIZE = 10000L; + /** + * 所有请求登录的code与channel关系 + */ + public static final Cache WAIT_LOGIN_MAP = Caffeine.newBuilder() + .expireAfterWrite(EXPIRE_TIME) + .maximumSize(MAX_MUM_SIZE) + .build(); /** * 所有已连接的websocket连接列表和一些额外参数 */ @@ -68,6 +74,7 @@ public class WebSocketServiceImpl implements WebSocketService { public static ConcurrentHashMap getOnlineMap() { return ONLINE_WS_MAP; } + /** * redis保存loginCode的key */ @@ -97,7 +104,7 @@ public class WebSocketServiceImpl implements WebSocketService { */ @SneakyThrows @Override - @FrequencyControl(time = 1000, count = 50, spEl = "T(com.abin.mallchat.common.common.utils.RequestHolder).get().getIp()") + @FrequencyControl(time = 1000, count = 50, spEl = "T(com.abin.mallchat.custom.user.websocket.NettyUtil).getAttr(#channel,T(com.abin.mallchat.custom.user.websocket.NettyUtil).IP)") public void handleLoginReq(Channel channel) { //生成随机不重复的登录码,并将channel存在本地cache中 Integer code = generateLoginCode(channel); @@ -114,13 +121,13 @@ public class WebSocketServiceImpl implements WebSocketService { * @return */ private Integer generateLoginCode(Channel channel) { - int inc = 0; + int inc; do { //本地cache时间必须比redis key过期时间短，否则会出现并发问题 - inc = RedisUtils.integerInc(RedisKey.getKey(LOGIN_CODE), 61, TimeUnit.MINUTES); - } while (CacheHolder.WAIT_LOGIN_MAP.asMap().containsKey(inc)); + inc = RedisUtils.integerInc(RedisKey.getKey(LOGIN_CODE), (int) EXPIRE_TIME.toMinutes(), TimeUnit.MINUTES); + } while (WAIT_LOGIN_MAP.asMap().containsKey(inc)); //储存一份在本地 - CacheHolder.WAIT_LOGIN_MAP.put(inc, channel); + WAIT_LOGIN_MAP.put(inc, channel); return inc; } @@ -130,7 +137,6 @@ public class WebSocketServiceImpl implements WebSocketService { * @param channel */ @Override -// @FrequencyControl(time = 10, count = 5, spEl = "T(com.abin.mallchat.common.common.utils.RequestHolder).get().getIp()") public void connect(Channel channel) { ONLINE_WS_MAP.put(channel, new WSChannelExtraDTO()); } @@ -207,30 +213,30 @@ public class WebSocketServiceImpl implements WebSocketService { } @Override - public Boolean scanLoginSuccess(Integer loginCode, User user, String token) { - //发送消息 - Channel channel = CacheHolder.WAIT_LOGIN_MAP.getIfPresent(loginCode); + public Boolean scanLoginSuccess(Integer loginCode, Long uid) { + //确认连接在该机器 + Channel channel = WAIT_LOGIN_MAP.getIfPresent(loginCode); if (Objects.isNull(channel)) { return Boolean.FALSE; } + User user = userDao.getById(uid); //移除code - CacheHolder.WAIT_LOGIN_MAP.invalidate(loginCode); + WAIT_LOGIN_MAP.invalidate(loginCode); + //调用用户登录模块 + String token = loginService.login(uid); //用户登录 loginSuccess(channel, user, token); - return true; + return Boolean.TRUE; } @Override - public Boolean scanSuccess(Integer loginCode, Long uid) { - Channel channel = CacheHolder.WAIT_LOGIN_MAP.getIfPresent(loginCode); + public Boolean scanSuccess(Integer loginCode) { + Channel channel = WAIT_LOGIN_MAP.getIfPresent(loginCode); if (Objects.nonNull(channel)) { sendMsg(channel, WSAdapter.buildScanSuccessResp()); return Boolean.TRUE; - }else { - //广播通知次channel服务扫码成功 - mqProducer.sendMsg(MQConstant.SCAN_MSG_TOPIC, new ScanSuccessMessageDTO(loginCode, WSAdapter.buildScanSuccessResp())); - return Boolean.FALSE; } + return Boolean.FALSE; } @@ -277,7 +283,7 @@ public class WebSocketServiceImpl implements WebSocketService { /** - * 给本地channel发送消息 + * 给本地channel发送消息 * * @param channel * @param wsBaseResp @@ -286,39 +292,4 @@ public class WebSocketServiceImpl implements WebSocketService { channel.writeAndFlush(new TextWebSocketFrame(JSONUtil.toJsonStr(wsBaseResp))); } - /** - * 案例证明ConcurrentHashMap#entrySet的值不是快照数据 - * - * @param args - * @throws InterruptedException - */ - public static void main(String[] args) throws InterruptedException { - ReentrantLock reentrantLock = new ReentrantLock(); - Condition condition = reentrantLock.newCondition(); - ConcurrentHashMap a = new ConcurrentHashMap<>(); - a.put(1, 1); - a.put(2, 2); - new Thread(() -> { - reentrantLock.lock(); - Set> entries = a.entrySet(); - System.out.println(entries); - try { - condition.await(); - } catch (InterruptedException e) { - e.printStackTrace(); - } - System.out.println(entries); - reentrantLock.unlock(); - - }).start(); - Thread.sleep(1000); - reentrantLock.lock(); - a.put(3, 3); - System.out.println("haha"); - condition.signalAll(); - reentrantLock.unlock(); - Thread.sleep(1000); - } - - } diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyCollectorHandler.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyCollectorHandler.java deleted file mode 100644 index 902bab7..0000000 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyCollectorHandler.java +++ /dev/null @@ -1,27 +0,0 @@ -package com.abin.mallchat.custom.user.websocket; - -import com.abin.mallchat.common.common.constant.MDCKey; -import com.abin.mallchat.common.common.domain.dto.RequestInfo; -import com.abin.mallchat.common.common.utils.RequestHolder; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import lombok.extern.slf4j.Slf4j; -import org.slf4j.MDC; - -import java.util.UUID; - - -@Slf4j -public class NettyCollectorHandler extends ChannelInboundHandlerAdapter { - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - String tid = UUID.randomUUID().toString(); - MDC.put(MDCKey.TID, tid); - RequestInfo info = new RequestInfo(); - info.setUid(NettyUtil.getAttr(ctx.channel(), NettyUtil.UID)); - info.setIp(NettyUtil.getAttr(ctx.channel(), NettyUtil.IP)); - RequestHolder.set(info); - - ctx.fireChannelRead(msg); - } -} diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServer.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServer.java index 4ebf470..4fca13b 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServer.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServer.java @@ -80,7 +80,6 @@ public class NettyWebSocketServer { pipeline.addLast(new HttpObjectAggregator(8192)); //保存用户ip pipeline.addLast(new HttpHeadersHandler()); - pipeline.addLast(new NettyCollectorHandler()); /** * 说明： * 1. 对于 WebSocket，它的数据是以帧frame 的形式传递的； diff --git a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServerHandler.java b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServerHandler.java index 8ca5ac6..7ff3d01 100644 --- a/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServerHandler.java +++ b/mallchat-custom-server/src/main/java/com/abin/mallchat/custom/user/websocket/NettyWebSocketServerHandler.java @@ -100,10 +100,6 @@ public class NettyWebSocketServerHandler extends SimpleChannelInboundHandler