啊啊啊好舒服视频,伊人开心网,欧美zozozo人禽交

主頁 > 知識庫 > 利用redis實現聊天記錄轉存功能的全過程

利用redis實現聊天記錄轉存功能的全過程

前言

前一陣子實現了我開源項目的單聊功能，在實現過程中遇到了需要將聊天記錄保存至數據庫的問題，在收到消息時肯定不能直接存數據庫，因為這樣在高并發的場景下，數據庫就炸了。

于是，我就想到了redis這個東西，第一次聽說它是在2年前，但是一直沒時間玩他，現在終于遇到了需要使用它的場景，在用的時候學它，本文就跟大家分享下我的實現思路以及過程，歡迎各位感興趣的開發者閱讀本文。

環境搭建

我的項目是基于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 RedisTemplateString, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplateString, 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 RedisTemplateObject, 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(MapString, Object> map) {
        redisTemplate.opsForValue().multiSet(map);
    }

    /**
     * 批量獲取值
     *
     * @param list 查詢的 Key 列表
     * @return value 列表
     */
    public ListObject> batchGet(ListString> 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, ListObject> value) {
        return redisTemplate.opsForList().leftPushAll(key, value);
    }

    /**
     * 向list末尾添加list數據
     *
     * @param key   鍵
     * @param value 值
     */
    public Long listRightPushAll(String key, ListObject> 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 第二個元素，依次類推；index0時，-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
        ListSubMessage> 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> ListT> castList(Object obj, ClassT> clazz) {
        ListT> 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();
    }
}

這里我設置的定時任務是從每月1號開始，每隔三天就執行一次，Quartz定時任務采用的是cron表達式，自己算這個比較麻煩，這里推薦一個在線網站，可以很容易的生成表達式：Cron表達式生成器