im-system 第十一章

在线状态功能

需求

需求一：需要实时的更新好友的状态，有一个标识可以辨别在线和离线，在线和离线可以实时得到感知，手动修改忙碌啥的状态可以实时通知到好友

需求二：打开群组等，可以获取到这一批人的在线状态，在线的会有一个在线的标识，和好友一样可以实时感知到用户下线了，可以实时的将在线修改为离线

改进

  1. 改进一：状态变更只推送给在线的用户
  2. 改进二：使用按需拉取、临时订阅的方式

NettyServerHandler#channelRead0&channelInactive

NettyServerHandler#channelRead0

//登录command
if (command == SystemCommand.LOGIN.getCommand()) {// 用户上线通知UserStatusChangeNotifyPack userStatusChangeNotifyPack = new UserStatusChangeNotifyPack();userStatusChangeNotifyPack.setAppId(msg.getMessageHeader().getAppId());userStatusChangeNotifyPack.setUserId(loginPack.getUserId());userStatusChangeNotifyPack.setStatus(ImConnectStatusEnum.ONLINE_STATUS.getCode());MqMessageProducer.sendMessage(userStatusChangeNotifyPack, msg.getMessageHeader(), UserEventCommand.USER_ONLINE_STATUS_CHANGE.getCommand());// 登录消息ackMessagePack<LoginAckPack> loginSuccess = new MessagePack<>();LoginAckPack loginAckPack = new LoginAckPack();loginSuccess.setAppId(msg.getMessageHeader().getAppId());loginSuccess.setImei(msg.getMessageHeader().getImei());loginAckPack.setUserId(loginPack.getUserId());loginSuccess.setData(loginAckPack);loginSuccess.setCommand(SystemCommand.LOGINACK.getCommand());ctx.channel().writeAndFlush(loginSuccess);
}

NettyServerHandler#channelInactive

//表示 channel 处于不活动状态
@Override
public void channelInactive(ChannelHandlerContext ctx) {//设置离线SessionSocketHolder.offlineUserSession((NioSocketChannel) ctx.channel());ctx.close();
}

MqMessageProducer#sendMessage

@Slf4j
public class MqMessageProducer {public static void sendMessage(Message message, Integer command) {Channel channel = null;String com = command.toString();// 获取指令 首字符, 指令类型String commandSub = com.substring(0, 1);CommandType commandType = CommandType.getCommandType(commandSub);String channelName = "";// 根据指令类型, 选择不同的交换机if (commandType == CommandType.MESSAGE) {// channelName为 pipeline2MessageServicechannelName = Constants.RabbitConstants.Im2MessageService;} else if (commandType == CommandType.GROUP) {// channelName为 pipeline2GroupServicechannelName = Constants.RabbitConstants.Im2GroupService;} else if (commandType == CommandType.FRIEND) {// channelName为 pipeline2FriendshipServicechannelName = Constants.RabbitConstants.Im2FriendshipService;} else if (commandType == CommandType.USER) {// channelName为 pipeline2UserServicechannelName = Constants.RabbitConstants.Im2UserService;}try {// 拿到对应channelName的 Channel 对象//（针对每个交换机 都有 1个 Channel对象 与之对应）channel = MqFactory.getChannel(channelName);JSONObject o = (JSONObject) JSON.toJSON(message.getMessagePack());o.put("command", command);o.put("clientType", message.getMessageHeader().getClientType());o.put("imei", message.getMessageHeader().getImei());o.put("appId", message.getMessageHeader().getAppId());// 用与交换机对应的channel对象 发送到名为 channelName 的交换机, 路由key为空字符串, 不指定属性, 消息内容//（这也就是说: tcp发送到rabbitmq中的上面4个交换机, 都使用针对每个交换机而创建的Channel对象,//            而逻辑层则使用了@RabbitListener注解来为这些个交换机绑定了各自的队列, 并消费消息）channel.basicPublish(channelName, "", null, o.toJSONString().getBytes());} catch (Exception e) {log.error("发送消息出现异常：{}", e.getMessage());}}public static void sendMessage(Object message, MessageHeader header, Integer command) {Channel channel = null;String com = command.toString();// 获取指令 首字符, 指令类型String commandSub = com.substring(0, 1);CommandType commandType = CommandType.getCommandType(commandSub);String channelName = "";if (commandType == CommandType.MESSAGE) {// channelName为 pipeline2MessageServicechannelName = Constants.RabbitConstants.Im2MessageService;} else if (commandType == CommandType.GROUP) {// channelName为 pipeline2GroupServicechannelName = Constants.RabbitConstants.Im2GroupService;} else if (commandType == CommandType.FRIEND) {// channelName为 pipeline2FriendshipServicechannelName = Constants.RabbitConstants.Im2FriendshipService;} else if (commandType == CommandType.USER) {// channelName为 pipeline2UserServicechannelName = Constants.RabbitConstants.Im2UserService;}try {// 拿到对应channelName的 Channel 对象channel = MqFactory.getChannel(channelName);JSONObject o = (JSONObject) JSON.toJSON(message);o.put("command", command);o.put("clientType", header.getClientType());o.put("imei", header.getImei());o.put("appId", header.getAppId());// 发送到名为 channelName 的交换机, 路由key为空字符串, 不指定属性, 消息内容//（这也就是说: tcp发送到rabbitmq中的上面4个交换机, 都使用各自创建的Channel对象,//            而逻辑层则使用了@RabbitListener注解来为这些个交换机绑定了各自的队列, 并消费消息）channel.basicPublish(channelName, "", null, o.toJSONString().getBytes());} catch (Exception e) {log.error("发送消息出现异常：{}", e.getMessage());}}}

UserOnlineStatusReceiver

@Component
public class UserOnlineStatusReceiver {private static Logger logger = LoggerFactory.getLogger(ChatOperateReceiver.class);@AutowiredImUserStatusService imUserStatusService;// 订阅MQ单聊消息队列--处理@RabbitListener(bindings = @QueueBinding(value = @Queue(value = Constants.RabbitConstants.Im2UserService, durable = "true"),exchange = @Exchange(value = Constants.RabbitConstants.Im2UserService, durable = "true")),concurrency = "1")public void onChatMessage(@Payload Message message,@Headers Map<String, Object> headers,Channel channel) throws Exception {long start = System.currentTimeMillis();Thread t = Thread.currentThread();String msg = new String(message.getBody(), "utf-8");logger.info("CHAT MSG FROM QUEUE :::::" + msg);//deliveryTag 用于回传 rabbitmq 确认该消息处理成功Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);try {JSONObject jsonObject = JSON.parseObject(msg);Integer command = jsonObject.getInteger("command");if (Objects.equals(command, UserEventCommand.USER_ONLINE_STATUS_CHANGE.getCommand())) {UserStatusChangeNotifyContent content = JSON.parseObject(msg, new TypeReference<UserStatusChangeNotifyContent>() {}.getType());//TODOimUserStatusService.processUserOnlineStatusNotify(content);}channel.basicAck(deliveryTag, false);} catch (Exception e) {logger.error("处理消息出现异常：{}", e.getMessage());logger.error("RMQ_CHAT_TRAN_ERROR", e);logger.error("NACK_MSG:{}", msg);//第一个false 表示不批量拒绝，第二个false表示不重回队列channel.basicNack(deliveryTag, false, false);} finally {long end = System.currentTimeMillis();logger.debug("channel {} basic-Ack ,it costs {} ms,threadName = {},threadId={}", channel, end - start, t.getName(), t.getId());}}
}

ImUserStatusServiceImpl

@Service
public class ImUserStatusServiceImpl implements ImUserStatusService {@AutowiredUserSessionUtils userSessionUtils;@AutowiredMessageProducer messageProducer;@AutowiredImFriendService imFriendService;@AutowiredStringRedisTemplate stringRedisTemplate;@Overridepublic void processUserOnlineStatusNotify(UserStatusChangeNotifyContent content) {// 获取在线的其它端sessionList<UserSession> userSessions = userSessionUtils.getUserSession(content.getAppId(), content.getUserId());UserStatusChangeNotifyPack userStatusChangeNotifyPack = new UserStatusChangeNotifyPack();BeanUtils.copyProperties(content, userStatusChangeNotifyPack);// 需要同步的其它端userStatusChangeNotifyPack.setClient(userSessions);// 同步给在线的其它端syncSender(userStatusChangeNotifyPack, content.getUserId(), content);// 通知其他人dispatcher(userStatusChangeNotifyPack, content.getUserId(), content.getAppId());}private void syncSender(Object pack, String userId, ClientInfo clientInfo) {// 同步给在线的其它端messageProducer.sendToUserExceptClient(userId,UserEventCommand.USER_ONLINE_STATUS_CHANGE_NOTIFY_SYNC,pack,clientInfo);}private void dispatcher(Object pack, String userId, Integer appId) {// 获取到用户的所有好友List<String> allFriendId = imFriendService.getAllFriendId(userId, appId);// 挨个给好友推送在线状态（有在线的就推, 没在线就没推了）for (String friendId : allFriendId) {messageProducer.sendToUser(friendId,UserEventCommand.USER_ONLINE_STATUS_CHANGE_NOTIFY,pack,appId);}// key的格式为: {appId}:subscribe:{userId}  value是个hashmap: 订阅的用户id -> 失效时间毫秒值String userKey = appId + ":" + Constants.RedisConstants.subscribe + userId;// 订阅该用户的所有用户idSet<Object> keys = stringRedisTemplate.opsForHash().keys(userKey);for (Object key : keys) {String filed = (String) key;Long expire = Long.valueOf((String) stringRedisTemplate.opsForHash().get(userKey, filed));// 如果没超过失效时间, 那就通知给订阅的用户if (expire > 0 && expire > System.currentTimeMillis()) {messageProducer.sendToUser(filed,UserEventCommand.USER_ONLINE_STATUS_CHANGE_NOTIFY,pack,appId);} else {// 如果超过失效时间, 那就删除stringRedisTemplate.opsForHash().delete(userKey, filed);}}}// 订阅用户在线状态@Overridepublic void subscribeUserOnlineStatus(SubscribeUserOnlineStatusReq req) {Long subExpireTime = 0L;if (req != null && req.getSubTime() > 0) {subExpireTime = System.currentTimeMillis() + req.getSubTime();}for (String beSubUserId : req.getSubUserId()) {// key的格式为: {appId}:subscribe:{userId}  value是个hashmap: 订阅的用户id -> 失效时间毫秒值String userKey = req.getAppId() + ":" + Constants.RedisConstants.subscribe + ":" + beSubUserId;stringRedisTemplate.opsForHash().put(userKey, req.getOperater(), subExpireTime.toString());}}// 设置用户自定义在线状态@Overridepublic void setUserCustomerStatus(SetUserCustomerStatusReq req) {UserCustomStatusChangeNotifyPack userCustomStatusChangeNotifyPack = new UserCustomStatusChangeNotifyPack();userCustomStatusChangeNotifyPack.setCustomStatus(req.getCustomStatus());userCustomStatusChangeNotifyPack.setCustomText(req.getCustomText());userCustomStatusChangeNotifyPack.setUserId(req.getUserId());// key的格式为: {appId}:userCustomerStatus:{userId}stringRedisTemplate.opsForValue().set(req.getAppId() + ":" + Constants.RedisConstants.userCustomerStatus + ":" + req.getUserId(),JSONObject.toJSONString(userCustomStatusChangeNotifyPack));// 同步给其它端syncSender(userCustomStatusChangeNotifyPack,req.getUserId(),new ClientInfo(req.getAppId(), req.getClientType(), req.getImei()));// 通知给好友和其它订阅当前用户状态的用户dispatcher(userCustomStatusChangeNotifyPack, req.getUserId(), req.getAppId());}@Overridepublic Map<String, UserOnlineStatusResp> queryFriendOnlineStatus(PullFriendOnlineStatusReq req) {// 获取用户的所有好友id列表List<String> allFriendId = imFriendService.getAllFriendId(req.getOperater(), req.getAppId());return getUserOnlineStatus(allFriendId, req.getAppId());}@Overridepublic Map<String, UserOnlineStatusResp> queryUserOnlineStatus(PullUserOnlineStatusReq req) {// 获取指定用户的在线状态return getUserOnlineStatus(req.getUserList(), req.getAppId());}private Map<String, UserOnlineStatusResp> getUserOnlineStatus(List<String> userIdList, Integer appId) {Map<String, UserOnlineStatusResp> result = new HashMap<>(userIdList.size());// 遍历所有的userIdfor (String userId : userIdList) {UserOnlineStatusResp resp = new UserOnlineStatusResp();// 获取指定userId对应的所有UserSessionList<UserSession> userSessions = userSessionUtils.getUserSession(appId, userId);resp.setSession(userSessions);// 格式为: {appId}:userCustomerStatus:{userId}String userKey = appId + ":" + Constants.RedisConstants.userCustomerStatus + ":" + userId;// 获取用户自定义状态String s = stringRedisTemplate.opsForValue().get(userKey);if (StringUtils.isNotBlank(s)) {JSONObject parse = (JSONObject) JSON.parse(s);resp.setCustomText(parse.getString("customText"));resp.setCustomStatus(parse.getInteger("customStatus"));}result.put(userId, resp);}return result;}}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/797839.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

vivado 使用网表插入调试探测流程

vivado 使用网表插入调试探测流程

使用网表插入调试探测流程 在 Vivado 工具中插入调试核的过程以分层方式来演示 &#xff0c; 以应对多样化的 Vivado 用户组的不同需求 &#xff1a; • 最高层是根据选定调试的一组信号线来自动创建并配置 Integrated Logic Analyzer (ILA) 核的简单向导。 • 下一层是“…
阅读更多...
File，IO流，递归详解

File，IO流，递归详解

File类 介绍 java.io.File类是Java语言提供了用来描述文件和目录(文件夹)的 构造 方法 注意&#xff1a; 构造方法中通常用的是第一个方法文件和目录可以通过File封装成对象File封装的对象仅仅是一个路径名&#xff0c;它是可以存在的&#xff0c;也可以不存在 绝对路径…
阅读更多...
【二分查找】Leetcode x 的平方根

【二分查找】Leetcode x 的平方根

题目解析 69. x 的平方根 这道题表面是没有顺序性&#xff0c;但是我们可以发现如果使用枚举策略&#xff0c;它是包含顺序性质的&#xff0c;因此我们将枚举算法改进成二分算法 算法讲解 我们让left指针指向1&#xff0c; right指针指向x 如果当前的mid计算出来的平方 &g…
阅读更多...
什么是商家转账到零钱

什么是商家转账到零钱

商家转账到零钱是什么&#xff1f; 通过商家转账到零钱这个功能&#xff0c;如果我们系统需要对用户支付费用&#xff0c;比如发放佣金、提成、退款之类的&#xff0c;可以直接转账到用户的微信零钱。 【商家转账到零钱】是【企业付款到零钱】的升级版&#xff0c;2022年5月1…
阅读更多...
怀俄明探空站数据解算PWV和Tm

怀俄明探空站数据解算PWV和Tm

1. Matlab 获取代码可关注公众号WZZHHH回复&#xff08;怀俄明探空站数据解算PWV和Tm&#xff09;&#xff0c;或者咸鱼关注&#xff1a;WZZHHH123 怀俄明探空站数据解算PWV和Tm&#xff1a; 有关 Matlab 获取代码可关注公众号WZZHHH回复&#xff08;怀俄明多线程下载&#…
阅读更多...
语音助手背后的魔法：一文揭秘NLP在智能助手中的应用

语音助手背后的魔法：一文揭秘NLP在智能助手中的应用

自然语言处理&#xff08;NLP&#xff09;正逐渐渗透到我们生活的每一个角落。在今天的大盘点中&#xff0c;我们将深入探索NLP在实际生活场景中的多样化应用&#xff0c;以及它在各个案例中所展现出的强大能力。 NLP技术如今已成为智能生活的关键驱动力&#xff0c;它让机器能…
阅读更多...
JavaScript笔记 11

JavaScript笔记 11

目录 01 创建元素的方式 02 BOM概述 03 window 04定时器 05 location对象的使用 07 js特效 08 offset系列相关属性 09 scroll 相关属性 10 client 相关属性 11 window 相关的事件 12 event 相关的属性 01 创建元素的方式 创建元素的三种方式: 1.innerHTML创建元素 …
阅读更多...
wordpress全站开发指南-面向开发者及深度用户（全中文实操）--php数组与基本循环

wordpress全站开发指南-面向开发者及深度用户（全中文实操）--php数组与基本循环

php数组与基本循环 <?php$myName"xixi";$namesarray(xixi1,xixi2,xixi3); ?> <p> Hi ,my name is <?php echo $myName; ?> </p> <p> Hi,my name is <?php echo $names[0] ?> </p> <p> Hi,my name is <?…
阅读更多...
Open CASCADE学习|放样建模

Open CASCADE学习|放样建模

在CAD软件中&#xff0c;Loft&#xff08;放样&#xff09;功能则是用于创建三维实体或曲面的重要工具。通过选取两个或多个横截面&#xff0c;并沿这些横截面进行放样&#xff0c;可以生成复杂的三维模型。在CAD放样功能的操作中&#xff0c;用户可以选择不同的选项来定制放样…
阅读更多...
创新数智化全场景福利解决方案，打造极致员工体验

创新数智化全场景福利解决方案，打造极致员工体验

众所周知&#xff0c;企业面临两个市场&#xff0c;一个是前端的产品&#xff08;服务&#xff09;市场&#xff0c;面对的是客户&#xff0c;另一个便是后端市场&#xff0c;即愈来愈烈的人才市场。在风云变幻、人潮涌动的知识经济时代&#xff0c;员工已成为企业未来的竞争关…
阅读更多...
洛谷 P4554 小明的游戏

洛谷 P4554 小明的游戏

思路&#xff1a;双端队列。 其实一开始你可以用BFS进行实验&#xff0c;由于我们需要找最小的费用&#xff0c;所以我们在BFS的时候可以这样想&#xff1a;在我们遍历到第一块板子的时候&#xff0c;在找周围的路时&#xff0c;我们可以改成这样的判断&#xff1a;如果周围的…
阅读更多...
【大数据】ETL工具适用场景、常见产品、功能介绍

【大数据】ETL工具适用场景、常见产品、功能介绍

ETL&#xff08;Extract, Transform, Load&#xff09;工具是用于将数据从源系统抽取出来、进行转换处理&#xff0c;然后加载到目标系统的软件工具。它们在数据仓库、数据湖、数据集成等领域起着至关重要的作用。以下是关于ETL工具的详细介绍&#xff1a; 适用场景&#xff1…
阅读更多...
Linux_进程通信_管道_system V共享内存_6

Linux_进程通信_管道_system V共享内存_6

文章目录 一、进程通信分类二、管道1.什么是管道1.原理2.管道的特点 2.匿名管道3.命名管道1.创建命名管道文件 - mkfifo (命令)2.创建命名管道文件 - mkfifo (函数) 三、system V共享内存1.原理2.共享内存函数1.fotk2.shmget1.如何知道有哪些IPC资源 - ipcs &#xff08;命令&a…
阅读更多...
河海大学-海洋学院2024年硕士研究生调剂通知

河海大学-海洋学院2024年硕士研究生调剂通知

一、调剂专业及计划具体调剂专业及计划可参见河海大学研究生院官网《河海大学2024年硕士研究生调剂通知》和附件。 二、调剂报名与复试要求 1.报名条件&#xff1a;调剂原则见《河海大学202 4年硕士研究生调剂通知》&#xff0c;详细要求见中国研究生招生信息网“全国硕士研究…
阅读更多...
Mysql密码修改问题

Mysql密码修改问题

docker安装mysql&#xff0c;直接拉取镜像&#xff0c;挂载关键目录即可启动&#xff0c;默认3306端口。此时无法直接连接&#xff0c;需要配置密码。docker进入mysql容器中 docker exec -it mysql bash #mysq是容器名称&#xff0c;也可以用容器id通过修改mysql的配置进行免密…
阅读更多...
Unity类银河恶魔城学习记录12-7-1 p129 Craft UI - part 1源代码

Unity类银河恶魔城学习记录12-7-1 p129 Craft UI - part 1源代码

Alex教程每一P的教程原代码加上我自己的理解初步理解写的注释&#xff0c;可供学习Alex教程的人参考 此代码仅为较上一P有所改变的代码 【Unity教程】从0编程制作类银河恶魔城游戏_哔哩哔哩_bilibili UI_CraftList.cs using System.Collections; using System.Collections.Gen…
阅读更多...
Vue基础知识：Vue路由——重定向，以及？（可选符）的使用

Vue基础知识：Vue路由——重定向，以及？（可选符）的使用

当网页打开时&#xff0c;url默认是/路径&#xff08;根路径&#xff09;&#xff0c;未匹配到组件时&#xff0c;会出现空白&#xff0c;为了解决这个问题 就要利用重定向 重定向&#xff1a;匹配path后&#xff0c;强制跳转path路径 重定向的语法&#xff1a; {path:匹配…
阅读更多...
【Linux】使用cloudreve搭建个人网盘并传输文件

【Linux】使用cloudreve搭建个人网盘并传输文件

Cloudreve 是一个开源的个人网盘系统&#xff0c;能够帮助用户搭建属于自己的私有云存储服务。它支持多种存储后端&#xff0c;包括本地存储、远程FTP/SFTP存储、以及云存储服务如阿里云OSS、腾讯云COS和Amazon S3等。Cloudreve具有友好的用户界面和丰富的功能&#xff0c;比如…
阅读更多...
bash工具-dir_util.sh

bash工具-dir_util.sh

#!/usr/bin/env bash#以pwd和当前脚本路径名$0 结合 给出 当前脚本所在目录名、当前脚本名 #调用者应该在切换目录之前调用本函数, 即 尽可能早的调用本脚本. # 若 调用者 切换到其他目录后&#xff0c;调用本脚本 则结果肯定不对. # 使用例子: getCurScriptDirName $0 #返…
阅读更多...
如何使用开源情报跟踪一个人？在线访问网站以及使用方法介绍

如何使用开源情报跟踪一个人？在线访问网站以及使用方法介绍

如何使用开源情报跟踪一个人&#xff1f;在线访问网站以及使用方法介绍。 开源情报&#xff08;OSINT&#xff09;是一门关于收集和分析公开可用信息的独特技艺&#xff0c;它致力于构建个人或团体的详尽档案。 这一过程中&#xff0c;信息搜集者会利用多元化的信息源&#xff…
阅读更多...
最新文章

Copyright @ 2022~2023