redis中怎么实现聊天记录转存功能
这篇文章将为大家详细讲解有关redis中怎么实现聊天记录转存功能,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。
环境搭建
我的项目是基于SpringBoot2.x搭建的,电脑已经安装了redis,用的maven作为jar包管理工具,所以只需要在maven中添加需要的依赖包即可,如果你用的是其他管理工具,请自行查阅如何添加依赖。
<!-- Redis --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!-- 定时任务调度 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-quartz</artifactId> <version>2.3.7.RELEASE</version> </dependency>
本文需要用到依赖:Redis 、quartz,在pom.xml文件的dependencies标签下添加下述代码。
spring: # redis配置 redis: host: 127.0.0.1 # redis地址 port: 6379 # 端口号 password: # 密码 timeout: 3000 # 连接超时时间,单位毫秒
实现思路
在websocket的服务中,收到客户端推送的消息后,我们对数据进行解析,构造聊天记录实体类,将其保存至redis中,最后我们使用quartz设置定时任务将redis的数据定时写入mysql中。
我们将上述思路进行下整理:
解析客户端数据,构造实体类
将数据保存至redis
使用quartz将redis中的数据定时写入mysql
实现过程
实现思路很简单,难在如何将实体类数据保存至redis,我们需要把redis这一块配置好后,才能继续实现我们的业务需求。
redis支持的数据结构类型有:
set 集合,string类型的无序集合,元素不允许重复
hash 哈希表,键值对的集合,用于存储对象
list 列表,链表结构
zset有序集合
string 字符串,最基本的数据类型,可以包含任何数据,比如一个序列化的对象,它的字符串大小上限是512MB
redis的客户端分为jedis 和 lettuce,在SpringBoot2.x中默认客户端是使用lettuce实现的,因此我们不用做过多配置,在使用的时候通过RedisTemplate.xxx来对redis进行操作即可。
自定义RedisTemplate
在RedisTemplate中,默认是使用Java字符串序列化,将字符串存入redis后可读性很差,因此,我们需要对他进行自定义,使用Jackson 序列化,以 JSON 方式进行存储。
我们在项目的config包下,创建一个名为LettuceRedisConfig的Java文件,我们再此文件中配置其默认序列化规则,它的代码如下:
package com.lk.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; // 自定义RedisTemplate设置序列化器, 方便转换redis中的数据与实体类互转 @Configuration public class LettuceRedisConfig { /** * Redis 序列化配置 */ @Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(connectionFactory); // 使用GenericJackson2JsonRedisSerializer替换默认序列化 GenericJackson2JsonRedisSerializer jackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer(); // 设置 Key 和 Value 的序列化规则 redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer); // 初始化 RedisTemplate 序列化完成 redisTemplate.afterPropertiesSet(); return redisTemplate; } }
封装redis工具类
做完上述操作后,通过RedisTemplate存储到redis中的数据就是json形式的了,接下来我们对其常用的操作封装成工具类,方便我们在项目中使用。
在Utils包中创建一个名为RedisOperatingUtil,其代码如下:
package com.lk.utils; import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @Component // Redis操作工具类 public class RedisOperatingUtil { @Resource private RedisTemplate<Object, Object> redisTemplate; /** * 指定 key 的过期时间 * * @param key 键 * @param time 时间(秒) */ public void setKeyTime(String key, long time) { redisTemplate.expire(key, time, TimeUnit.SECONDS); } /** * 根据 key 获取过期时间(-1 即为永不过期) * * @param key 键 * @return 过期时间 */ public Long getKeyTime(String key) { return redisTemplate.getExpire(key, TimeUnit.SECONDS); } /** * 判断 key 是否存在 * * @param key 键 * @return 如果存在 key 则返回 true,否则返回 false */ public Boolean hasKey(String key) { return redisTemplate.hasKey(key); } /** * 删除 key * * @param key 键 */ public Long delKey(String... key) { if (key == null || key.length < 1) { return 0L; } return redisTemplate.delete(Arrays.asList(key)); } /** * 获取 Key 的类型 * * @param key 键 */ public String keyType(String key) { DataType dataType = redisTemplate.type(key); assert dataType != null; return dataType.code(); } /** * 批量设置值 * * @param map 要插入的 key value 集合 */ public void barchSet(Map<String, Object> map) { redisTemplate.opsForValue().multiSet(map); } /** * 批量获取值 * * @param list 查询的 Key 列表 * @return value 列表 */ public List<Object> batchGet(List<String> list) { return redisTemplate.opsForValue().multiGet(Collections.singleton(list)); } /** * 获取指定对象类型key的值 * * @param key 键 * @return 值 */ public Object objectGetKey(String key) { return redisTemplate.opsForValue().get(key); } /** * 设置对象类型的数据 * * @param key 键 * @param value 值 */ public void objectSetValue(String key, Object value) { redisTemplate.opsForValue().set(key, value); } /** * 向list的头部插入一条数据 * * @param key 键 * @param value 值 */ public Long listLeftPush(String key, Object value) { return redisTemplate.opsForList().leftPush(key, value); } /** * 向list的末尾插入一条数据 * * @param key 键 * @param value 值 */ public Long listRightPush(String key, Object value) { return redisTemplate.opsForList().rightPush(key, value); } /** * 向list头部添加list数据 * * @param key 键 * @param value 值 */ public Long listLeftPushAll(String key, List<Object> value) { return redisTemplate.opsForList().leftPushAll(key, value); } /** * 向list末尾添加list数据 * * @param key 键 * @param value 值 */ public Long listRightPushAll(String key, List<Object> value) { return redisTemplate.opsForList().rightPushAll(key, value); } /** * 通过索引设置list元素的值 * * @param key 键 * @param index 索引 * @param value 值 */ public void listIndexSet(String key, long index, Object value) { redisTemplate.opsForList().set(key, index, value); } /** * 获取列表指定范围内的list元素,正数则表示正向查找,负数则倒叙查找 * * @param key 键 * @param start 开始 * @param end 结束 * @return boolean */ public Object listRange(String key, long start, long end) { return redisTemplate.opsForList().range(key, start, end); } /** * 从列表前端开始取出数据 * * @param key 键 * @return 结果数组对象 */ public Object listPopLeftKey(String key) { return redisTemplate.opsForList().leftPop(key); } /** * 从列表末尾开始遍历取出数据 * * @param key 键 * @return 结果数组 */ public Object listPopRightKey(String key) { return redisTemplate.opsForList().rightPop(key); } /** * 获取list长度 * * @param key 键 * @return 列表长度 */ public Long listLen(String key) { return redisTemplate.opsForList().size(key); } /** * 通过索引获取list中的元素 * * @param key 键 * @param index 索引(index>=0时,0 表头,1 第二个元素,依次类推;index<0时,-1,表尾,-2倒数第二个元素,依次类推) * @return 列表中的元素 */ public Object listIndex(String key, long index) { return redisTemplate.opsForList().index(key, index); } /** * 移除list元素 * * @param key 键 * @param count 移除数量("负数"则从列表倒叙查找删除 count 个对应的值; "整数"则从列表正序查找删除 count 个对应的值;) * @param value 值 * @return 成功移除的个数 */ public Long listRem(String key, long count, Object value) { return redisTemplate.opsForList().remove(key, count, value); } /** * 截取指定范围内的数据, 移除不是范围内的数据 * @param key 操作的key * @param start 截取开始位置 * @param end 截取激素位置 */ public void listTrim(String key, long start, long end) { redisTemplate.opsForList().trim(key, start, end); } }
进行单元测试
做完上述操作后,最难弄的一关我们就已经搞定了,接下来我们来对一会需要使用的方法进行单元测试,确保其能够正常运行。
创建一个名为RedisTest的Java文件,注入需要用到的相关类。
redisOperatingUtil为我们的redis工具类
subMessageMapper为聊天记录表的dao层
@RunWith(SpringRunner.class) @SpringBootTest @Slf4j public class RedisTest { @Resource private RedisOperatingUtil redisOperatingUtil; @Resource private SubMessageMapper subMessageMapper; }
接下来,我们看下SubMessage实体类的代码。
package com.lk.entity; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; @Getter @Setter @NoArgsConstructor @AllArgsConstructor // 聊天记录-消息内容 public class SubMessage { private Integer id; private String msgText; // 消息内容 private String createTime; // 创建时间 private String userName; // 用户名 private String userId; // 推送方用户id private String avatarSrc; // 推送方头像 private String msgId; // 接收方用户id private Boolean status; // 消息状态 }
测试list数据的写入与获取
在单元测试类内部加入下述代码:
@Test public void testSerializableListRedisTemplate() { // 构造聊天记录实体类数据 SubMessage subMessage = new SubMessage(); subMessage.setAvatarSrc("https://www.kaisir.cn/uploads/1ece3749801d4d45933ba8b31403c685touxiang.jpeg"); subMessage.setUserId("1090192"); subMessage.setUserName("神奇的程序员"); subMessage.setMsgText("你好"); subMessage.setMsgId("2901872"); subMessage.setCreateTime("2020-12-12 18:54:06"); subMessage.setStatus(false); // 将聊天记录对象保存到redis中 redisOperatingUtil.listRightPush("subMessage", subMessage); // 获取list中的数据 Object resultObj = redisOperatingUtil.listRange("subMessage", 0, redisOperatingUtil.listLen("subMessage")); // 将Object安全的转为List List<SubMessage> resultList = ObjectToOtherUtil.castList(resultObj, SubMessage.class); // 遍历获取到的结果 if (resultList != null) { for (SubMessage message : resultList) { System.out.println(message.getUserName()); } } }
在上述代码中,我们从redis中取出的数据是Object类型的,我们要将它转换为与之对应的实体类,一开始我是用的类型强转,但是idea会报黄色警告,于是就写了一个工具类用于将Object对象安全的转换为与之对应的类型,代码如下:
package com.lk.utils; import java.util.ArrayList; import java.util.List; public class ObjectToOtherUtil { public static <T> List<T> castList(Object obj, Class<T> clazz) { List<T> result = new ArrayList<>(); if (obj instanceof List<?>) { for (Object o : (List<?>) obj) { result.add(clazz.cast(o)); } return result; } return null; } }
执行后,我们看看redis是否有保存到我们写入的数据,如下所示,已经成功保存。
我们再来看看,代码的执行结果,看看有没有成功获取到数据,如下图所示,也成功取到了。
注意:如果你的项目对websocket进行了启动配置,可能会导致单元测试失败,报错java.lang.IllegalStateException: Failed to load ApplicationContext,解决方案就是注释掉websocket配置文件中的@Configuration即可。
测试list数据的取出
当我们把redis中存储的数据迁移到mysql后,需要删除redis中的数据,一开始我用的是它的delete方法,但是他的delete方法只能删除与之匹配的值,不能选择一个区间进行删除,于是就决定用它的pop方法进行出栈操作。
我们来测试下工具类中的listPopLeftKey方法。
@Test public void testListPop() { long item = 0; // 获取存储在redis中聊天记录的条数 long messageListSize = redisOperatingUtil.listLen("subMessage"); for (int i = 0; i < messageListSize; i++) { // 从头向尾取出链表中的元素 SubMessage messageResult = (SubMessage) redisOperatingUtil.listPopLeftKey("subMessage"); log.info(messageResult.getMsgText()); item++; } log.info(item+"条数据已成功取出"); }
执行结果如下所示,成功取出了redis中存储的两条数据。
测试聊天记录转移至数据库
接下来我们在redis中放入三条数据用于测试
我们测试下将redis中的数据取出,然后写入数据库,代码如下:
// 测试聊天记录转移数据库 @Test public void testRedisToMysqlTask() { // 获取存储在redis中聊天记录的条数 long messageListSize = redisOperatingUtil.listLen("subMessage"); // 写入数据库的数据总条数 long resultCount = 0; for (int i = 0; i < messageListSize; i++) { // 从头到尾取出链表中的元素 SubMessage subMessage= (SubMessage) redisOperatingUtil.listPopLeftKey("subMessage"); // 向数据库写入数据 int result = subMessageMapper.addMessageTextInfo(subMessage); if (result > 0) { // 写入成功 resultCount++; } } log.info(resultCount+ "条聊天记录,已写入数据库"); }
执行结果如下,数据已成功写入数据库且redis中的数据也被删除。
解析客户端数据保存至redis
完成上述操作后,我们redis那一块的东西就搞定了,接下来就可以实现将客户端的数据存到redis里了。
这里有个坑,因为websocket服务类中用到了@Component,会导致redis的工具类注入失败,出现null的情况,解决这个问题需要将当前类名声明为静态变量,然后在init中获取赋值redis工具类,代码如下:
// 解决redis操作工具类注入为null的问题 public static WebSocketServer webSocketServer; @PostConstruct public void init() { webSocketServer = this; webSocketServer.redisOperatingUtil = this.redisOperatingUtil; }
在websocket服务的@OnMessage注解中,收到客户端发送的消息,我们将其保存到redis中,代码如下:
/** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息 * // @param session 客户端会话 */ @OnMessage public void onMessage(String message) { // 客户端发送的消息 JSONObject jsReply = new JSONObject(message); // 添加在线人数 jsReply.put("onlineUsers", getOnlineCount()); if (jsReply.has("buddyId")) { // 获取推送方id String userId = jsReply.getString("userID"); // 获取被推送方id String buddyId = jsReply.getString("buddyId"); // 非测试数据则推送消息 if (!buddyId.equals("121710f399b84322bdecc238199d6888")) { // 发送消息至推送方 this.sendInfo(jsReply.toString(), userId); } // 构造聊天记录实体类数据 SubMessage subMessage = new SubMessage(); subMessage.setAvatarSrc(jsReply.getString("avatarSrc")); subMessage.setUserId(jsReply.getString("userID")); subMessage.setUserName(jsReply.getString("username")); subMessage.setMsgText(jsReply.getString("msg")); subMessage.setMsgId(jsReply.getString("msgId")); subMessage.setCreateTime(DateUtil.getThisTime()); subMessage.setStatus(false); // 将聊天记录对象保存到redis中 webSocketServer.redisOperatingUtil.listRightPush("subMessage", subMessage); // 发送消息至被推送方 this.sendInfo(jsReply.toString(), buddyId); } }
做完上述操作后,收到客户端发送的消息就会自动写入redis。
定时将redis的数据写入mysql
接下来,我们使用quartz定时向mysql中写入数据,他执行定时任务的步骤分为2步:
创建任务类编写任务内容
在QuartzConfig文件中设置定时,执行第一步创建的任务。
首先,创建quartzServer包,在其下创建RedisToMysqlTask.java文件,在此文件内实现redis写入mysql的代码
package com.lk.quartzServer; import com.lk.dao.SubMessageMapper; import com.lk.entity.SubMessage; import com.lk.utils.RedisOperatingUtil; import lombok.extern.slf4j.Slf4j; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; import org.springframework.scheduling.quartz.QuartzJobBean; import javax.annotation.Resource; // 将redis数据放进mysql中 @Slf4j public class RedisToMysqlTask extends QuartzJobBean { @Resource private RedisOperatingUtil redisOperatingUtil; @Resource private SubMessageMapper subMessageMapper; @Override protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException { // 获取存储在redis中聊天记录的条数 long messageListSize = redisOperatingUtil.listLen("subMessage"); // 写入数据库的数据总条数 long resultCount = 0; for (int i = 0; i < messageListSize; i++) { // 从头到尾取出链表中的元素 SubMessage subMessage= (SubMessage) redisOperatingUtil.listPopLeftKey("subMessage"); // 向数据库写入数据 int result = subMessageMapper.addMessageTextInfo(subMessage); if (result > 0) { // 写入成功 resultCount++; } } log.info(resultCount+ "条聊天记录,已写入数据库"); } }
在config包下创建QuartzConfig.java文件,创建定时任务
package com.lk.config; import com.lk.quartzServer.RedisToMysqlTask; import org.quartz.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * Quartz定时任务配置 */ @Configuration public class QuartzConfig { @Bean public JobDetail RedisToMysqlQuartz() { // 执行定时任务 return JobBuilder.newJob(RedisToMysqlTask.class).withIdentity("CallPayQuartzTask").storeDurably().build(); } @Bean public Trigger CallPayQuartzTaskTrigger() { //cron方式,从每月1号开始,每隔三天就执行一次 return TriggerBuilder.newTrigger().forJob(RedisToMysqlQuartz()) .withIdentity("CallPayQuartzTask") .withSchedule(CronScheduleBuilder.cronSchedule("* * 4 1/3 * ?")) .build(); } }
关于redis中怎么实现聊天记录转存功能就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。