关键词搜索

源码搜索 ×
×

Java基于队列和数据库批量维护用户在线离线状态和位置信息

发布2020-08-29浏览801次

详情内容

目录

前言介绍

用户数据迁移

用户状态数据批量处理

队列数据批量消费

前言介绍

系统用户实时位置都保存在redis中，我们采用redis过期方式来监听用户位置离线，离线的用户key会放入对应的队列进行消费（这里的队列可以是Java队列也可以是RabbitMQ之类的消息中间件，我们采用的是ConcurrentLinkedQueue）。项目中对用户实时位置有两个要求（特殊行业对人员位置安全规范的要求）：

  • a.用户最后一次经纬度实时位置更新到数据库
  • b.用户离线后更新用户在线状态

为了解决每次用户位置心跳更新导致数据库连接不足问题，我采用了批量更新的方式来解决大批量的用户上线和离线更新。解决了如下问题：

  • a.解决用户状态表统计数据不一致问题
  • b.解决数据库更新占用数据库连接过多问题(可实现批量更新上线状态位置和离线状态，最后一次位置保存更新降低数据库压力)
  • c.解决redis和database在线状态延迟过大问题

用户数据迁移

针对需要维护最后用户在线位置数据的这类用户，我们有单独用户角色加以限定。这种用户只需要简单将数据做个分表保存起来就可以了。首次初始化这类数据是根据已有用户数据直接采用SQL 支持的SELECT [A,B,....] INTO TABLE_SUB FROM TABLE_MAIN 

示例脚本如下：

  1. -- 注意：======依赖触发器更新用户信息=====
  2. --  修改目的：
  3. --  a.解决用户状态表格统计数据不一致问题
  4. --  b.解决数据库更新占用数据库连接过多问题(可实现批量更新上线状态位置和离线状态，最后一次位置保存更新降低数据库压力)
  5. --  c.解决redis和database在线状态延迟过大问题
  6. -- 可反复执行如下表数据记录
  7. DROP TABLE if exists xh_yw.xh_user_online_tb ;
  8.  
  9. SELECT
  10.   -- 组织机构ID
  11.   i_orgid,
  12.   -- 组织机构编号
  13. 	c_orgbh,
  14. 	-- 组织机构名称
  15. 	c_orgname,
  16. 	-- 用户ID
  17. 	i_userid,
  18. 	-- 用户姓名
  19. 	c_userealname,
  20. 	-- 手机号码
  21. 	c_usertel,
  22. 	-- 时间戳
  23.   make_timestamp(2020,1,1,0,0,0) AS lasttime,
  24.   -- 最后位置:经度
  25.   0.0 AS longitude,
  26.   -- 最后位置:纬度
  27. 	0.0 AS latitude,
  28. 	-- 是否在线: 0 离线  1 在线
  29. 	0 AS is_online ,
  30. 	-- 日期make_date(2020,1,1) AS last_date
  31. 	'2020-01-01' AS last_date
  32.  
  33. INTO xh_yw.xh_user_online_tb
  34.  
  35. FROM xh_ht.fs_yw_base_user where i_userid
  36. -- 角色过滤
  37. IN (select  distinct i_userid from xh_ht.fs_yw_user_role where i_roleid = 5 )

用户状态数据批量处理

批量处理都依赖数据库支持的方式。

MySQL参考：https://www.cnblogs.com/mslagee/p/6509682.html

Postgresql参考：https://www.itranslater.com/qa/detailshttps://files.jxasp.com/image/2583251656280376320

Java代码示例：

  1. package com.patrol.position.service;
  2.  
  3. import com.alibaba.fastjson.JSONArray;
  4. import com.forestar.platform.dao.DatabaseRepository;
  5. import com.patrol.beans.Constants;
  6. import com.patrol.beans.user.UserPosition;
  7. import com.patrol.beans.util.LogicUtil;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.apache.commons.lang3.ObjectUtils;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.beans.factory.annotation.Qualifier;
  12. import org.springframework.data.redis.core.RedisTemplate;
  13. import org.springframework.jdbc.core.JdbcTemplate;
  14. import org.springframework.stereotype.Service;
  15. import java.text.SimpleDateFormat;
  16. import java.util.*;
  17.  
  18. /**
  19.  * @Copyright: 2019-2021
  20.  * @FileName: UserOnlineService.java
  21.  * @Author: PJL
  22.  * @Date: 2020/7/15 19:37
  23.  * @Description: 用户在线中间数据表服务【通过队列方式批量更新】
  24.  */
  25. @Slf4j
  26. @Service
  27. public class UserOnlineService {
  28.  
  29.     /**
  30.      * Redis查询工具模板类
  31.      */
  32.     @Qualifier("redisTemplateByLettuce")
  33.     @Autowired
  34.     RedisTemplate redisTemplate;
  35.  
  36.     @Autowired
  37.     JdbcTemplate jdbcTemplate;
  38.  
  39.     @Autowired
  40.     DatabaseRepository databaseRepository;
  41.  
  42.     SimpleDateFormat sdfTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  43.  
  44.     SimpleDateFormat sdfDate = new SimpleDateFormat("yyyy-MM-dd");
  45.  
  46.     /**
  47.      * 用户在线位置中间表
  48.      */
  49.     final String XH_USER_ONLINE_TB = "XH_USER_ONLINE_TB";
  50.  
  51.     /**
  52.      * 解析临时表数据多列值
  53.      *
  54.      * @param list
  55.      * @return
  56.      */
  57.     private List parseMultiParams(List<UserPosition> list) {
  58.         List result = new ArrayList();
  59.         int count = list.size();
  60.         Integer[] userArray = new Integer[count];
  61.         String[] lastTimeArray = new String[count];
  62.         String[] lastDateArray = new String[count];
  63.         Double[] longitudeArray = new Double[count];
  64.         Double[] latitudeArray = new Double[count];
  65.         UserPosition userPosition;
  66.         Date date;
  67.         for (int i = 0; i < count; i++) {
  68.             userPosition = list.get(i);
  69.             if (null != userPosition) {
  70.                 date = new Date(userPosition.getTimestamp());
  71.                 userArray[i] = Integer.valueOf(userPosition.getUserId());
  72.                 lastTimeArray[i] = new StringBuffer("date_trunc('second', TIMESTAMP '").append(sdfTime.format(date)).append("')").toString();
  73.                 String[] dateValues = sdfDate.format(date).split("-");
  74.                 lastDateArray[i] = new StringBuffer("make_date(").append(dateValues[0]).append(",").append(dateValues[1]).append(",").append(dateValues[2]).append(")").toString();
  75.                 longitudeArray[i] = userPosition.getPosition()[0];
  76.                 latitudeArray[i] = userPosition.getPosition()[1];
  77.             }
  78.         }
  79.         result.add(userArray);
  80.         result.add(lastTimeArray);
  81.         result.add(lastDateArray);
  82.         result.add(longitudeArray);
  83.         result.add(latitudeArray);
  84.         return result;
  85.     }
  86.  
  87.  
  88.     /**
  89.      * 批量更新用户上线状态表
  90.      *
  91.      * @param list
  92.      */
  93.     public void batchOnline(List<UserPosition> list) {
  94.         if (ObjectUtils.isNotEmpty(list)) {
  95.             List paramList = this.parseMultiParams(list);
  96.             String userIds = JSONArray.toJSONString(paramList.get(0));
  97.             StringBuffer lastTimes = new StringBuffer();
  98.             String[] timeList = (String[]) paramList.get(1);
  99.             for (String s : timeList) {
  100.                 if (lastTimes.length() == 0) {
  101.                     lastTimes.append(s);
  102.                 } else {
  103.                     lastTimes.append(",").append(s);
  104.                 }
  105.             }
  106.             String[] dateList = (String[]) paramList.get(2);
  107.             StringBuffer lastDates = new StringBuffer();
  108.             for (String s : dateList) {
  109.                 if (lastDates.length() == 0) {
  110.                     lastDates.append(s);
  111.                 } else {
  112.                     lastDates.append(",").append(s);
  113.                 }
  114.             }
  115.             String longitudes = JSONArray.toJSONString(paramList.get(3));
  116.             String latitudes = JSONArray.toJSONString(paramList.get(4));
  117.             StringBuffer sb = new StringBuffer(" UPDATE ")
  118.                     .append(Constants.DB_YW_TABLE_SPACE).append(XH_USER_ONLINE_TB).append(" a ")
  119.                     .append(" SET ")
  120.                     .append(" LASTTIME = u.LASTTIME,")
  121.                     .append(" LAST_DATE = u.LAST_DATE,")
  122.                     .append(" LONGITUDE = u.LONGITUDE,")
  123.                     .append(" LATITUDE = u.LATITUDE,")
  124.                     .append(" IS_ONLINE = 1 ")
  125.                     .append(" FROM ( SELECT ")
  126.                     .append(" unnest(array").append(userIds).append(") ").append(" as I_USERID,")
  127.                     .append(" unnest(array[").append(lastTimes).append("]) ").append(" as LASTTIME,")
  128.                     .append(" unnest(array[").append(lastDates).append("]) ").append(" as LAST_DATE,")
  129.                     .append(" unnest(array").append(longitudes).append(") ").append(" as LONGITUDE,")
  130.                     .append(" unnest(array").append(latitudes).append(") ").append(" as LATITUDE")
  131.                     .append("  ) as u ")
  132.                     .append(" WHERE a.I_USERID = u.I_USERID ");
  133.             jdbcTemplate.execute(sb.toString());
  134.         }
  135.  
  136.     }
  137.  
  138.     /**
  139.      * 用户离线状态修改(0：离线  1:在线)
  140.      *
  141.      * @param userIdList
  142.      */
  143.     public void updateUserOffline(List<String> userIdList) {
  144.         String[] userIds = new String[userIdList.size()];
  145.         userIds = userIdList.toArray(userIds);
  146.         String filter = LogicUtil.getOrgFilterString(userIds);
  147.         String sql = new StringBuffer(" UPDATE ").append(XH_USER_ONLINE_TB).append(" SET IS_ONLINE = 0 WHERE I_USERID IN (").append(filter).append(")").toString();
  148.         databaseRepository.execute(XH_USER_ONLINE_TB, sql);
  149.     }
  150. }

注意：Postgresql函数日期和时间戳需要特殊处理。

队列数据批量消费

用户在线离线只需要两个队列就可以区分开处理了。

  1. package com.patrol.position.queue;
  2.  
  3. import com.patrol.beans.user.UserPosition;
  4. import com.patrol.config.condition.ServerCondition;
  5. import com.patrol.position.service.UserOnlineService;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.commons.lang3.ObjectUtils;
  8. import org.apache.commons.lang3.StringUtils;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import javax.annotation.PostConstruct;
  13. import java.util.ArrayList;
  14. import java.util.List;
  15. import java.util.concurrent.ConcurrentLinkedQueue;
  16.  
  17. /**
  18.  * @Copyright: 2019-2021
  19.  * @FileName: UpdateStatusOnlineQueue.java
  20.  * @Author: PJL
  21.  * @Date: 2020/8https://files.jxasp.com/image/20 11:03
  22.  * @Description: 在线离线状态列表数据更新队列
  23.  */
  24. @Slf4j
  25. @Component
  26. public class UpdateStatusQueue {
  27.  
  28.     @Autowired
  29.     UserOnlineService userOnlineService;
  30.  
  31.     /**
  32.      * 并发链表队列--在线位置队列
  33.      */
  34.     private static final ConcurrentLinkedQueue<UserPosition> onlineQueue = new ConcurrentLinkedQueue<>();
  35.  
  36.     /**
  37.      * 并发链表队列--离线位置队列
  38.      */
  39.     private static final ConcurrentLinkedQueue<String> offlineQueue = new ConcurrentLinkedQueue<>();
  40.  
  41.     /**
  42.      * 消费用户上线、离线下线任务
  43.      */
  44.     @PostConstruct
  45.     private void consumeUserOnlineStatusQueue() {
  46.         if (ServerCondition.isServer) {
  47.             log.info(">>>>>>>>启动服务端消费线程....");
  48.             /***********用户上线批量消费***********/
  49.             this.userOnline();
  50.  
  51.             /***********用户离线状态批量消费***********/
  52.             this.userOffline();
  53.  
  54.             log.info(">>>>>>>>启动服务端消费线程....完毕!");
  55.         }
  56.     }
  57.  
  58.  
  59.     /**
  60.      * 添加在线用户
  61.      *
  62.      * @param userPosition
  63.      */
  64.     public static void addToOnlineQueue(UserPosition userPosition) {
  65.         if (ServerCondition.isServer) {
  66.             onlineQueue.add(userPosition);
  67.         }
  68.     }
  69.  
  70.     /**
  71.      * 添加离线用户
  72.      *
  73.      * @param userId
  74.      */
  75.     public static void addToOfflineQueue(String userId) {
  76.         if (ServerCondition.isServer) {
  77.             offlineQueue.add(userId);
  78.         }
  79.     }
  80.  
  81.     /**
  82.      * 用户上线更新最后一次上线位置批量处理
  83.      */
  84.     private void userOnline() {
  85.         new Thread(new Runnable() {
  86.             @Override
  87.             public void run() {
  88.                 while (true) {
  89.                     int size = onlineQueue.size();
  90.                     if (size > 0) {
  91.                         List<UserPosition> list = new ArrayList<>();
  92.                         UserPosition userPosition;
  93.                         for (int i = 0; i < size; i++) {
  94.                             userPosition = onlineQueue.poll();
  95.                             if (null != userPosition) {
  96.                                 list.add(userPosition);
  97.                             }
  98.                         }
  99.                         if (ObjectUtils.isNotEmpty(list)) {
  100.                             // 批量上线
  101.                             userOnlineService.batchOnline(list);
  102.                         }
  103.                     }
  104.                     try {
  105.                         Thread.sleep(50);
  106.                     } catch (InterruptedException e) {
  107.                         e.printStackTrace();
  108.                     }
  109.                 }
  110.             }
  111.         }).start();
  112.     }
  113.  
  114.     /**
  115.      * 更新用户在线状态表为离线
  116.      */
  117.     private void userOffline() {
  118.         new Thread(new Runnable() {
  119.             @Override
  120.             public void run() {
  121.                 while (true) {
  122.                     int size = offlineQueue.size();
  123.                     if (size > 0) {
  124.                         List<String> list = new ArrayList<>();
  125.                         String userId;
  126.                         for (int i = 0; i < size; i++) {
  127.                             userId = offlineQueue.poll();
  128.                             if (null != userId && StringUtils.isNotEmpty(userId)) {
  129.                                 list.add(userId);
  130.                             }
  131.                         }
  132.                         if (ObjectUtils.isNotEmpty(list)) {
  133.                             // 批量离线
  134.                             userOnlineService.updateUserOffline(list);
  135.                         }
  136.                     }
  137.                     try {
  138.                         Thread.sleep(50);
  139.                     } catch (InterruptedException e) {
  140.                         e.printStackTrace();
  141.                     }
  142.                 }
  143.             }
  144.         }).start();
  145.     }
  146. }

这里开了两个线程处理，根据用户规模可以改为多线程并发消费（但是请注意控制数据库连接）。

 

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载