Files
backend/internal/cache/cache.go
lan 90c57f1a1c
All checks were successful
Build Backend / build (push) Successful in 2m7s
Build Backend / build-docker (push) Successful in 1m51s
feat(core): optimize performance and reliability through batching and redis-backed unread counts
This commit introduces several significant architectural improvements to enhance system performance, scalability, and reliability:

- **Performance Optimization (N+1 Problem Resolution)**: Implemented batching mechanisms across `MessageHandler`, `ChatService`, `GroupService`, `MessageService`, and `UserService`. This replaces multiple individual database/cache queries with single batch operations for fetching unread counts, last messages, participants, and user information.
- **Enhanced Unread Count Management**: Migrated unread count tracking to a Redis Hash-based approach (`unread#️⃣{userID}`). This allows for atomic increments/decrements and efficient retrieval of both individual conversation unread counts and total unread counts for a user.
- **Improved Message Sequencing**: Integrated Redis-based sequence (`seq`) pre-allocation for messages to ensure strict ordering and reduce database contention during high-concurrency message creation.
- **Push Service Reliability**: Refactored the push notification system to include persistent `PushRecord` tracking. Added a recovery mechanism (`recoverPendingPushes`) to reload pending notifications from the database upon service startup, ensuring better delivery guarantees.
- **WebSocket Reliability**: Updated the WebSocket hub to move away from in-memory history replay in favor of a client-driven synchronization model (`sync_required` event and `ack` handling), reducing memory overhead and improving connection stability.
- **Cache Layer Enhancements**: Added `HIncrBy` and `IncrBySeq` to the `Cache` interface and its implementations (`RedisCache`, `LayeredCache`) to support the new unread and sequence management logic.
2026-05-04 18:31:03 +08:00

270 lines
6.9 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package cache
import (
"cmp"
"context"
"encoding/json"
"errors"
"math/rand"
"sync"
"time"
"github.com/redis/go-redis/v9"
)
var ErrKeyNotFound = errors.New("key not found")
// Cache 缓存接口
type Cache interface {
// Set 设置缓存值支持TTL
Set(key string, value any, ttl time.Duration)
// Get 获取缓存值
Get(key string) (any, bool)
// Delete 删除缓存
Delete(key string)
// DeleteByPrefix 根据前缀删除缓存
DeleteByPrefix(prefix string)
// Clear 清空所有缓存
Clear()
// Exists 检查键是否存在
Exists(key string) bool
// Increment 增加计数器的值
Increment(key string) int64
// IncrementBy 增加指定值
IncrementBy(key string, value int64) int64
// DeleteBatch 批量删除多个 key使用 Pipeline 减少网络往返)
DeleteBatch(keys []string)
// ==================== Hash 操作 ====================
// HSet 设置 Hash 字段
HSet(ctx context.Context, key string, field string, value any) error
// HMSet 批量设置 Hash 字段
HMSet(ctx context.Context, key string, values map[string]any) error
// HGet 获取 Hash 字段值
HGet(ctx context.Context, key string, field string) (string, error)
// HMGet 批量获取 Hash 字段值
HMGet(ctx context.Context, key string, fields ...string) ([]any, error)
// HGetAll 获取 Hash 所有字段
HGetAll(ctx context.Context, key string) (map[string]string, error)
// HDel 删除 Hash 字段
HDel(ctx context.Context, key string, fields ...string) error
// HIncrBy 原子递增 Hash 字段的值
HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error)
// ==================== Sorted Set 操作 ====================
// ZAdd 添加 Sorted Set 成员
ZAdd(ctx context.Context, key string, score float64, member string) error
// ZRangeByScore 按分数范围获取成员(升序)
ZRangeByScore(ctx context.Context, key string, min, max string, offset, count int64) ([]string, error)
// ZRevRangeByScore 按分数范围获取成员(降序)
ZRevRangeByScore(ctx context.Context, key string, max, min string, offset, count int64) ([]string, error)
// ZRevRange 按排名倒序取成员score 从高到低start/stop 为排名下标,含端点)
ZRevRange(ctx context.Context, key string, start, stop int64) ([]string, error)
// ZReplaceSortedSet 原子替换整个 ZSET先删后批量写入
ZReplaceSortedSet(ctx context.Context, key string, members []redis.Z) error
// ZRem 删除 Sorted Set 成员
ZRem(ctx context.Context, key string, members ...any) error
// ZCard 获取 Sorted Set 成员数量
ZCard(ctx context.Context, key string) (int64, error)
// ==================== 计数器操作 ====================
// Incr 原子递增(返回新值)
Incr(ctx context.Context, key string) (int64, error)
// IncrBySeq 原子递增指定值(用于 seq 对齐)
IncrBySeq(ctx context.Context, key string, delta int64) (int64, error)
// Expire 设置过期时间
Expire(ctx context.Context, key string, ttl time.Duration) error
}
const nullMarkerValue = "__carrot_cache_null__"
var loadLocks sync.Map
type Settings struct {
Enabled bool
KeyPrefix string
DefaultTTL time.Duration
NullTTL time.Duration
JitterRatio float64
PostListTTL time.Duration
ConversationTTL time.Duration
UnreadCountTTL time.Duration
GroupMembersTTL time.Duration
DisableFlushDB bool
}
var settings = Settings{
Enabled: true,
DefaultTTL: 30 * time.Second,
NullTTL: 5 * time.Second,
JitterRatio: 0.1,
PostListTTL: 30 * time.Second,
ConversationTTL: 120 * time.Second,
UnreadCountTTL: 30 * time.Second,
GroupMembersTTL: 120 * time.Second,
DisableFlushDB: true,
}
func Configure(s Settings) {
settings.Enabled = s.Enabled
if s.KeyPrefix != "" {
settings.KeyPrefix = s.KeyPrefix
}
if s.DefaultTTL > 0 {
settings.DefaultTTL = s.DefaultTTL
}
if s.NullTTL > 0 {
settings.NullTTL = s.NullTTL
}
if s.JitterRatio > 0 {
settings.JitterRatio = s.JitterRatio
}
if s.PostListTTL > 0 {
settings.PostListTTL = s.PostListTTL
}
if s.ConversationTTL > 0 {
settings.ConversationTTL = s.ConversationTTL
}
if s.UnreadCountTTL > 0 {
settings.UnreadCountTTL = s.UnreadCountTTL
}
if s.GroupMembersTTL > 0 {
settings.GroupMembersTTL = s.GroupMembersTTL
}
settings.DisableFlushDB = s.DisableFlushDB
}
func GetSettings() Settings {
return settings
}
func normalizeKey(key string) string {
return cmp.Or(settings.KeyPrefix, "") + ":" + key
}
// ResolveKey 返回带全局前缀的实际 Redis 键(供需直连 Redis 的组件使用)
func ResolveKey(key string) string {
return normalizeKey(key)
}
func normalizePrefix(prefix string) string {
return cmp.Or(settings.KeyPrefix, "") + ":" + prefix
}
func SetWithJitter(c Cache, key string, value any, ttl time.Duration, jitterRatio float64) {
if !settings.Enabled {
return
}
c.Set(key, value, ApplyTTLJitter(ttl, jitterRatio))
}
func SetNull(c Cache, key string, ttl time.Duration) {
if !settings.Enabled {
return
}
c.Set(key, nullMarkerValue, ttl)
}
func ApplyTTLJitter(ttl time.Duration, jitterRatio float64) time.Duration {
if ttl <= 0 || jitterRatio <= 0 {
return ttl
}
if jitterRatio > 1 {
jitterRatio = 1
}
maxJitter := int64(float64(ttl) * jitterRatio)
if maxJitter <= 0 {
return ttl
}
delta := rand.Int63n(maxJitter + 1)
return ttl + time.Duration(delta)
}
func GetTyped[T any](c Cache, key string) (T, bool) {
var zero T
if !settings.Enabled {
return zero, false
}
raw, ok := c.Get(key)
if !ok {
recordMiss()
return zero, false
}
if str, ok := raw.(string); ok && str == nullMarkerValue {
recordHit()
return zero, false
}
if typed, ok := raw.(T); ok {
recordHit()
return typed, true
}
var out T
switch v := raw.(type) {
case string:
if err := json.Unmarshal([]byte(v), &out); err != nil {
recordDecodeError()
return zero, false
}
recordHit()
return out, true
case []byte:
if err := json.Unmarshal(v, &out); err != nil {
recordDecodeError()
return zero, false
}
recordHit()
return out, true
default:
data, err := json.Marshal(v)
if err != nil {
recordDecodeError()
return zero, false
}
if err := json.Unmarshal(data, &out); err != nil {
recordDecodeError()
return zero, false
}
recordHit()
return out, true
}
}
func GetOrLoadTyped[T any](
c Cache,
key string,
ttl time.Duration,
jitterRatio float64,
nullTTL time.Duration,
loader func() (T, error),
) (T, error) {
if cached, ok := GetTyped[T](c, key); ok {
return cached, nil
}
lockValue, _ := loadLocks.LoadOrStore(key, &sync.Mutex{})
lock := lockValue.(*sync.Mutex)
lock.Lock()
defer lock.Unlock()
if cached, ok := GetTyped[T](c, key); ok {
return cached, nil
}
loaded, err := loader()
if err != nil {
var zero T
return zero, err
}
encoded, marshalErr := json.Marshal(loaded)
if marshalErr == nil && string(encoded) == "null" && nullTTL > 0 {
SetNull(c, key, nullTTL)
return loaded, nil
}
SetWithJitter(c, key, loaded, ttl, jitterRatio)
return loaded, nil
}