Files
backend/internal/cache/layered_cache.go

462 lines
9.7 KiB
Go
Raw Permalink Normal View History

package cache
import (
"context"
"strconv"
"sync"
"time"
redislib "github.com/redis/go-redis/v9"
redisPkg "with_you/internal/pkg/redis"
)
type LayeredCache struct {
local *LRUStriped
redis *RedisCache
mu sync.RWMutex
stats CacheStats
enabled bool
keyPrefix string
}
type CacheStats struct {
LocalHits int64
RedisHits int64
Misses int64
Sets int64
Invalidates int64
}
type LayeredCacheOptions struct {
LocalSize int
LocalBuckets int
LocalDefaultTTL time.Duration
KeyPrefix string
Enabled bool
}
func NewLayeredCache(redisClient *redisPkg.Client, opts *LayeredCacheOptions) *LayeredCache {
if opts == nil {
opts = &LayeredCacheOptions{
LocalSize: 10000,
LocalBuckets: 0,
LocalDefaultTTL: 5 * time.Minute,
KeyPrefix: "",
Enabled: true,
}
}
if opts.LocalSize <= 0 {
opts.LocalSize = 10000
}
if opts.LocalDefaultTTL <= 0 {
opts.LocalDefaultTTL = 5 * time.Minute
}
localCache := NewLRUStriped(&LRUStripedOptions{
Size: opts.LocalSize,
DefaultExpiry: opts.LocalDefaultTTL,
Buckets: opts.LocalBuckets,
Name: "layered_local",
})
var redisCache *RedisCache
if redisClient != nil {
redisCache = NewRedisCache(redisClient)
}
return &LayeredCache{
local: localCache,
redis: redisCache,
enabled: opts.Enabled,
keyPrefix: opts.KeyPrefix,
}
}
func (c *LayeredCache) Set(key string, value any, ttl time.Duration) {
if !c.enabled {
return
}
c.mu.Lock()
c.stats.Sets++
c.mu.Unlock()
c.local.Set(key, value, ttl)
if c.redis != nil {
c.redis.Set(key, value, ttl)
}
}
func (c *LayeredCache) Get(key string) (any, bool) {
if !c.enabled {
return nil, false
}
if val, ok := c.local.Get(key); ok {
c.mu.Lock()
c.stats.LocalHits++
c.mu.Unlock()
return val, true
}
if c.redis != nil {
if val, ok := c.redis.Get(key); ok {
c.mu.Lock()
c.stats.RedisHits++
c.mu.Unlock()
if ttl := c.getLocalTTL(); ttl > 0 {
refactor: improve system stability, performance, and code structure This commit introduces several architectural improvements and optimizations across the codebase: - **Performance & Reliability**: - Implemented Redis pipelining in `ConversationCache.CacheMessage` to reduce network round-trips. - Added a circuit breaker to the JPush client to prevent cascading failures. - Introduced batch deletion and batch member addition capabilities in repositories. - Added message idempotency support using `client_msg_id` and a Redis-based cache. - Optimized WebSocket handling with connection limits (total and per-user) and improved error logging. - **Code Refactoring**: - Refactored `Router` to use a `RouterDeps` struct, simplifying the constructor and improving maintainability. - Unified model ID generation logic using new `id_helper.go` (supporting UUID and Snowflake). - Standardized JSON serialization/deserialization in models using `json_helper.go`. - Refactored DTO conversion logic, specifically for `UserResponse` (using functional options) and `Report` responses. - Removed redundant/deprecated DTOs like `PostDetailResponse` and `TradeItemDetailResponse`. - **Cache Improvements**: - Enhanced `LayeredCache` with `SetRaw` to avoid double-encoding when promoting values from Redis to local cache. - Added `DeleteBatch` support to the cache interface. - **Other Changes**: - Cleaned up `config.go` by removing redundant default values and explicit environment variable overrides. - Improved WebSocket registration flow to handle connection limits gracefully.
2026-05-04 13:07:03 +08:00
if str, ok := val.(string); ok {
c.local.SetRaw(key, []byte(str), ttl)
} else {
c.local.Set(key, val, ttl)
}
}
return val, true
}
}
c.mu.Lock()
c.stats.Misses++
c.mu.Unlock()
return nil, false
}
func (c *LayeredCache) getLocalTTL() time.Duration {
return 30 * time.Second
}
func (c *LayeredCache) Delete(key string) {
if !c.enabled {
return
}
c.mu.Lock()
c.stats.Invalidates++
c.mu.Unlock()
c.local.Delete(key)
if c.redis != nil {
c.redis.Delete(key)
}
}
func (c *LayeredCache) DeleteByPrefix(prefix string) {
if !c.enabled {
return
}
c.mu.Lock()
c.stats.Invalidates++
c.mu.Unlock()
c.local.DeleteByPrefix(prefix)
if c.redis != nil {
c.redis.DeleteByPrefix(prefix)
}
}
refactor: improve system stability, performance, and code structure This commit introduces several architectural improvements and optimizations across the codebase: - **Performance & Reliability**: - Implemented Redis pipelining in `ConversationCache.CacheMessage` to reduce network round-trips. - Added a circuit breaker to the JPush client to prevent cascading failures. - Introduced batch deletion and batch member addition capabilities in repositories. - Added message idempotency support using `client_msg_id` and a Redis-based cache. - Optimized WebSocket handling with connection limits (total and per-user) and improved error logging. - **Code Refactoring**: - Refactored `Router` to use a `RouterDeps` struct, simplifying the constructor and improving maintainability. - Unified model ID generation logic using new `id_helper.go` (supporting UUID and Snowflake). - Standardized JSON serialization/deserialization in models using `json_helper.go`. - Refactored DTO conversion logic, specifically for `UserResponse` (using functional options) and `Report` responses. - Removed redundant/deprecated DTOs like `PostDetailResponse` and `TradeItemDetailResponse`. - **Cache Improvements**: - Enhanced `LayeredCache` with `SetRaw` to avoid double-encoding when promoting values from Redis to local cache. - Added `DeleteBatch` support to the cache interface. - **Other Changes**: - Cleaned up `config.go` by removing redundant default values and explicit environment variable overrides. - Improved WebSocket registration flow to handle connection limits gracefully.
2026-05-04 13:07:03 +08:00
func (c *LayeredCache) DeleteBatch(keys []string) {
if !c.enabled || len(keys) == 0 {
return
}
c.mu.Lock()
c.stats.Invalidates += int64(len(keys))
c.mu.Unlock()
for _, k := range keys {
c.local.Delete(k)
}
if c.redis != nil {
c.redis.DeleteBatch(keys)
}
}
func (c *LayeredCache) Clear() {
c.local.Clear()
if c.redis != nil {
c.redis.Clear()
}
c.mu.Lock()
c.stats = CacheStats{}
c.mu.Unlock()
}
func (c *LayeredCache) Exists(key string) bool {
if !c.enabled {
return false
}
if c.local.Exists(key) {
return true
}
if c.redis != nil {
return c.redis.Exists(key)
}
return false
}
func (c *LayeredCache) Increment(key string) int64 {
if !c.enabled {
return 0
}
if c.redis != nil {
val := c.redis.Increment(key)
c.local.Delete(key)
return val
}
return c.local.Increment(key)
}
func (c *LayeredCache) IncrementBy(key string, value int64) int64 {
if !c.enabled {
return 0
}
if c.redis != nil {
val := c.redis.IncrementBy(key, value)
c.local.Delete(key)
return val
}
return c.local.IncrementBy(key, value)
}
func (c *LayeredCache) GetStats() CacheStats {
c.mu.RLock()
defer c.mu.RUnlock()
return c.stats
}
func (c *LayeredCache) GetHitRate() float64 {
c.mu.RLock()
defer c.mu.RUnlock()
total := c.stats.LocalHits + c.stats.RedisHits + c.stats.Misses
if total == 0 {
return 0
}
return float64(c.stats.LocalHits+c.stats.RedisHits) / float64(total)
}
func (c *LayeredCache) HSet(ctx context.Context, key string, field string, value any) error {
if !c.enabled || c.redis == nil {
return nil
}
c.local.Delete(key)
return c.redis.HSet(ctx, key, field, value)
}
func (c *LayeredCache) HMSet(ctx context.Context, key string, values map[string]any) error {
if !c.enabled || c.redis == nil {
return nil
}
c.local.Delete(key)
return c.redis.HMSet(ctx, key, values)
}
func (c *LayeredCache) HGet(ctx context.Context, key string, field string) (string, error) {
if !c.enabled || c.redis == nil {
return "", ErrKeyNotFound
}
return c.redis.HGet(ctx, key, field)
}
func (c *LayeredCache) HMGet(ctx context.Context, key string, fields ...string) ([]any, error) {
if !c.enabled || c.redis == nil {
return nil, ErrKeyNotFound
}
return c.redis.HMGet(ctx, key, fields...)
}
func (c *LayeredCache) HGetAll(ctx context.Context, key string) (map[string]string, error) {
if !c.enabled || c.redis == nil {
return nil, ErrKeyNotFound
}
return c.redis.HGetAll(ctx, key)
}
func (c *LayeredCache) HDel(ctx context.Context, key string, fields ...string) error {
if !c.enabled || c.redis == nil {
return nil
}
c.local.Delete(key)
return c.redis.HDel(ctx, key, fields...)
}
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:hash:{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
func (c *LayeredCache) HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error) {
if !c.enabled || c.redis == nil {
return 0, ErrKeyNotFound
}
c.local.Delete(key)
return c.redis.HIncrBy(ctx, key, field, incr)
}
func (c *LayeredCache) ZAdd(ctx context.Context, key string, score float64, member string) error {
if !c.enabled || c.redis == nil {
return nil
}
c.local.Delete(key)
return c.redis.ZAdd(ctx, key, score, member)
}
func (c *LayeredCache) ZRangeByScore(ctx context.Context, key string, min, max string, offset, count int64) ([]string, error) {
if !c.enabled || c.redis == nil {
return nil, ErrKeyNotFound
}
return c.redis.ZRangeByScore(ctx, key, min, max, offset, count)
}
func (c *LayeredCache) ZRevRangeByScore(ctx context.Context, key string, max, min string, offset, count int64) ([]string, error) {
if !c.enabled || c.redis == nil {
return nil, ErrKeyNotFound
}
return c.redis.ZRevRangeByScore(ctx, key, max, min, offset, count)
}
func (c *LayeredCache) ZRevRange(ctx context.Context, key string, start, stop int64) ([]string, error) {
if !c.enabled || c.redis == nil {
return nil, ErrKeyNotFound
}
return c.redis.ZRevRange(ctx, key, start, stop)
}
func (c *LayeredCache) ZReplaceSortedSet(ctx context.Context, key string, members []redislib.Z) error {
if !c.enabled || c.redis == nil {
return ErrKeyNotFound
}
c.local.Delete(key)
return c.redis.ZReplaceSortedSet(ctx, key, members)
}
func (c *LayeredCache) ZRem(ctx context.Context, key string, members ...any) error {
if !c.enabled || c.redis == nil {
return nil
}
c.local.Delete(key)
return c.redis.ZRem(ctx, key, members...)
}
func (c *LayeredCache) ZCard(ctx context.Context, key string) (int64, error) {
if !c.enabled || c.redis == nil {
return 0, ErrKeyNotFound
}
return c.redis.ZCard(ctx, key)
}
// IsEnabled 返回 Redis 是否可用
func (c *LayeredCache) IsEnabled() bool {
return c.enabled && c.redis != nil
}
func (c *LayeredCache) Incr(ctx context.Context, key string) (int64, error) {
if !c.IsEnabled() {
return 0, ErrKeyNotFound
}
c.local.Delete(key)
return c.redis.Incr(ctx, key)
}
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:hash:{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
func (c *LayeredCache) IncrBySeq(ctx context.Context, key string, delta int64) (int64, error) {
if !c.enabled || c.redis == nil {
return 0, ErrKeyNotFound
}
c.local.Delete(key)
return c.redis.IncrBySeq(ctx, key, delta)
}
func (c *LayeredCache) Expire(ctx context.Context, key string, ttl time.Duration) error {
if !c.enabled || c.redis == nil {
return nil
}
return c.redis.Expire(ctx, key, ttl)
}
func (c *LayeredCache) SetLocalOnly(key string, value any, ttl time.Duration) {
if !c.enabled {
return
}
c.local.Set(key, value, ttl)
}
func (c *LayeredCache) GetLocalOnly(key string) (any, bool) {
if !c.enabled {
return nil, false
}
return c.local.Get(key)
}
func (c *LayeredCache) SetRedisOnly(key string, value any, ttl time.Duration) {
if !c.enabled || c.redis == nil {
return
}
c.redis.Set(key, value, ttl)
}
func (c *LayeredCache) GetRedisOnly(key string) (any, bool) {
if !c.enabled || c.redis == nil {
return nil, false
}
return c.redis.Get(key)
}
func (c *LayeredCache) GetRedisClient() *redisPkg.Client {
if c.redis == nil {
return nil
}
return c.redis.GetRedisClient()
}
func (c *LayeredCache) InvalidateUserCache(userID uint) {
c.DeleteByPrefix(cacheKeyPrefixUser + ":")
c.Delete(buildUserDetailKey(userID))
}
func (c *LayeredCache) InvalidatePostCache(postID uint) {
c.Delete(buildPostDetailKey(postID))
c.DeleteByPrefix(cacheKeyPrefixPostList)
}
func (c *LayeredCache) InvalidateGroupCache(groupID uint) {
c.Delete(buildGroupDetailKey(groupID))
c.Delete(buildGroupMembersKey(groupID))
}
func (c *LayeredCache) InvalidateAllLists() {
c.DeleteByPrefix(cacheKeyPrefixPostList)
c.DeleteByPrefix(cacheKeyPrefixUserActivity)
}
var (
cacheKeyPrefixUser = "user"
cacheKeyPrefixPostList = "post_list"
cacheKeyPrefixUserActivity = "user_activity"
)
func buildUserDetailKey(userID uint) string {
return "user:detail:" + strconv.FormatUint(uint64(userID), 10)
}
func buildPostDetailKey(postID uint) string {
return "post:detail:" + strconv.FormatUint(uint64(postID), 10)
}
func buildGroupDetailKey(groupID uint) string {
return "group:detail:" + strconv.FormatUint(uint64(groupID), 10)
}
func buildGroupMembersKey(groupID uint) string {
return "group:members:" + strconv.FormatUint(uint64(groupID), 10)
}
var _ Cache = (*LayeredCache)(nil)