Merge pull request #117 from po-lan/main

对缓存进一步优化

.gitignore

@@ -15,3 +15,4 @@ dist/
 # Build
 *.db
 checkVersion.sh
+bun.lockb

src/common/utils/LRUCache.ts

@@ -21,7 +21,8 @@ class cacheNode<T> {
   }
 }
 
-type cache<T> = { [key: group_id]: { [key: user_id]: cacheNode<T> } };
+type cache<T, K = { [key: user_id]: cacheNode<T> }> = { [key: group_id]: K };
+type removeObject<T> = cache<T, { userId: user_id, value: T }[]>
 class LRU<T> {
   private maxAge: number;
   private maxSize: number;
@@ -29,9 +30,9 @@ class LRU<T> {
   private cache: cache<T>;
   private head: cacheNode<T> | null = null;
   private tail: cacheNode<T> | null = null;
-  private onFuncs: ((node: cacheNode<T>) => void)[] = [];
+  private onFuncs: ((node: removeObject<T>) => void)[] = [];
 
-  constructor(maxAge: number = 6e4, maxSize: number = 5e3) {
+  constructor(maxAge: number = 6e4 * 3, maxSize: number = 1e4) {
     this.maxAge = maxAge;
     this.maxSize = maxSize;
     this.cache = Object.create(null);
@@ -53,46 +54,39 @@ class LRU<T> {
     node.prev = node.next = null;
     delete this.cache[node.groupId][node.userId];
     this.removeNode(node);
-    this.onFuncs.forEach((func) => func(node));
+    this.onFuncs.forEach((func) => func({ [node.groupId]: [node] }));
     this.currentSize--;
   }
 
-  public on(func: (node: cacheNode<T>) => void) {
+  public on(func: (node: removeObject<T>) => void) {
     this.onFuncs.push(func);
   }
 
   private removeExpired() {
     const now = Date.now();
     let current = this.tail;
-    const nodesToRemove: cacheNode<T>[] = [];
+    let totalNodeNum = 0;
-    let removedCount = 0;
+
+    const removeObject: cache<T, { userId: user_id, value: T }[]> = {};
 
-    // 收集需要删除的节点
     while (current && now - current.timestamp > this.maxAge) {
-      nodesToRemove.push(current);
+      // 收集节点
+      if (!removeObject[current.groupId]) removeObject[current.groupId] = [];
+      removeObject[current.groupId].push({ userId: current.userId, value: current.value });
+      // 删除LRU节点
+      delete this.cache[current.groupId][current.userId]
       current = current.prev;
-      removedCount++;
+      totalNodeNum++;
-      if (removedCount >= 100) break;
+      this.currentSize--
     }
 
-    // 更新链表指向
+    if (!totalNodeNum) return;
-    if (nodesToRemove.length > 0) {
-      const newTail = nodesToRemove[nodesToRemove.length - 1].prev;
-      if (newTail) {
-        newTail.next = null;
-      } else {
-        this.head = null;
-      }
-      this.tail = newTail;
-    }
 
-    nodesToRemove.forEach((node) => {
+    // 跟新链表指向
-      node.prev = node.next = null;
+    if (current) { current.next = null } else { this.head = null }
-      delete this.cache[node.groupId][node.userId];
+    this.tail = current
 
-      this.currentSize--;
+    this.onFuncs.forEach(func => func(removeObject))
-      this.onFuncs.forEach((func) => func(node));
-    });
   }
 
   private addNode(node: cacheNode<T>) {
@@ -144,7 +138,7 @@ class LRU<T> {
   public get(groupId: group_id, userId: user_id): null | { userId: user_id; value: T };
   public get(groupId: group_id, userId?: user_id): any {
     const groupObject = this.cache[groupId];
-    if(!groupObject) return userId === undefined ? [] : null;
+    if (!groupObject) return userId === undefined ? [] : null;
 
     if (userId === undefined) {
       return Object.entries(groupObject).map(([userId, { value }]) => ({
@@ -162,4 +156,6 @@ class LRU<T> {
   }
 }
 
+
+
 export default LRU;

src/common/utils/db.ts

@@ -72,7 +72,7 @@ class DBUtil extends DBUtilBase {
     private cache: { gid: number; uid: number }[] = [];
     private maxSize: number;
 
-    constructor(maxSize: number = 5000) {
+    constructor(maxSize: number = 50000) {
       this.maxSize = maxSize;
     }
 
@@ -120,57 +120,83 @@ class DBUtil extends DBUtilBase {
     });
 
 
-    this.LURCache.on(async (node) => {
+    this.LURCache.on(async (nodeObject) => {
-      const { value: time, groupId, userId } = node;
 
-      logDebug('插入发言时间', userId, groupId);
+      Object.entries(nodeObject).forEach(async ([_groupId, datas]) => {
-      await this.createGroupInfoTimeTableIfNotExist(groupId);
+        const userIds = datas.map(v => v.userId);
+        const groupId = Number(_groupId)
+        logDebug('插入发言时间', _groupId);
 
-      const method = await this.getDataSetMethod(groupId, userId);
+        await this.createGroupInfoTimeTableIfNotExist(groupId);
-      logDebug('插入发言时间方法判断', userId, groupId, method);
 
-      const sql =
+        const needCreatUsers = await this.getNeedCreatList(groupId, userIds);
-        method == 'update'
+        const updateList = needCreatUsers.length > 0 ? datas.filter(user => !needCreatUsers.includes(user.userId)) : datas;
-          ? `UPDATE "${groupId}" SET last_sent_time = ? WHERE user_id = ?`
+        const insertList = needCreatUsers.map(userId => datas.find(e => userId == e.userId)!)
-          : `INSERT INTO "${groupId}" (last_sent_time, user_id)  VALUES (?, ?)`;
 
-      this.db!.all(sql, [time, userId], (err) => {
+        logDebug(`updateList`, updateList);
-        if (err) {
+        logDebug(`insertList`, insertList)
-          return logError('插入/更新发言时间失败', userId, groupId);
+
+        if (insertList.length) {
+          const insertSql = `INSERT INTO "${groupId}" (last_sent_time, user_id) VALUES ${insertList.map(() => '(?, ?)').join(', ')};`
+
+          this.db!.all(insertSql, insertList.map(v => [v.value, v.userId]).flat(), err => {
+            if (err) {
+              logError(`群 ${groupId} 插入失败`)
+              logError(`更新Sql : ${insertSql}`)
+            }
+          })
         }
-        logDebug('插入/更新发言时间成功', userId, groupId);
+
-      });
+        if (updateList.length) {
+          const updateSql =
+            `UPDATE "${groupId}" SET last_sent_time = CASE ` +
+            updateList.map(v => `WHEN user_id = ${v.userId} THEN ${v.value}`).join(" ") +
+            " ELSE last_sent_time END WHERE user_id IN " +
+            `(${updateList.map(v => v.userId).join(", ")});`
+
+          this.db!.all(updateSql, [], err => {
+            if (err) {
+              logError(`群 ${groupId} 跟新失败`)
+              logError(`更新Sql : ${updateSql}`)
+            }
+          })
+        }
+
+      })
+
 
     });
   }
-  async getDataSetMethod(groupId: number, userId: number) {
+  async getNeedCreatList(groupId: number, userIds: number[]) {
-    // 缓存记录
+
-    if (this.LastSentCache.get(groupId, userId)) {
+    // 获取缓存中没有的
-      logDebug('缓存命中', userId, groupId);
+    const unhas = userIds.filter(userId => !this.LastSentCache.get(groupId, userId));
-      return 'update';
+
+    if (unhas.length == 0) {
+      logDebug('缓存全部命中');
+      return [];
     }
 
-    // 数据库判断
+    logDebug('缓存未全部命中');
-    return new Promise<'insert' | 'update'>((resolve, reject) => {
-      this.db!.all(
-        `SELECT * FROM "${groupId}" WHERE user_id = ?`,
-        [userId],
-        (err, rows) => {
-          if (err) {
-            logError('查询发言时间存在失败', userId, groupId, err);
-            return logError('插入发言时间失败', userId, groupId, err);
-          }
 
-          if (rows.length === 0) {
+    const sql = `SELECT * FROM "${groupId}" WHERE user_id IN (${unhas.map(() => '?').join(',')})`;
-            logDebug('查询发言时间不存在', userId, groupId);
-            return resolve('insert');
-          }
 
-          logDebug('查询发言时间存在', userId, groupId);
+    return new Promise<number[]>((resolve) => {
-          resolve('update');
+      this.db!.all(sql, unhas, (err, rows: { user_id: number }[]) => {
+        const has = rows.map(v => v.user_id);
+        const needCreatUsers = unhas.filter(userId => !has.includes(userId));
+
+        if (needCreatUsers.length == 0) {
+          logDebug('数据库全部命中')
+        } else {
+          logDebug('数据库未全部命中')
         }
-      );
+
-    });
+        resolve(needCreatUsers)
+      })
+    })
+
+
   }
   async createGroupInfoTimeTableIfNotExist(groupId: number) {
     const createTableSQL = (groupId: number) =>
@@ -408,12 +434,12 @@ class DBUtil extends DBUtilBase {
     logDebug('读取发言时间', groupId);
     return new Promise<IRember[]>((resolve, reject) => {
       this.db!.all(`SELECT * FROM "${groupId}" `, (err, rows: IRember[]) => {
-        const cache = this.LURCache.get(groupId).map(e=>({user_id:e.userId, last_sent_time:e.value}));
+        const cache = this.LURCache.get(groupId).map(e => ({ user_id: e.userId, last_sent_time: e.value }));
         if (err) {
           logError('查询发言时间失败', groupId);
-          return resolve(cache.map(e=>({...e, join_time:0})));
+          return resolve(cache.map(e => ({ ...e, join_time: 0 })));
         }
-         Object.assign(rows, cache)
+        Object.assign(rows, cache)
         logDebug('查询发言时间成功', groupId, rows);
         resolve(rows);
       });
@@ -439,8 +465,8 @@ class DBUtil extends DBUtilBase {
       (err) => {
         if (err)
           logError(err),
-          Promise.reject(),
+            Promise.reject(),
-          logError('插入入群时间失败', userId, groupId);
+            logError('插入入群时间失败', userId, groupId);
       }
     );