refactor(P3): Redis 缓存键版本化 — Client 增加 Key() 方法，默认前缀 v1:
This commit is contained in:
Sisyphus
@@ -1,156 +1,155 @@
|
|||||||
package account
|
package account
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"feedsystem_video_go/internal/auth"
|
"feedsystem_video_go/internal/auth"
|
||||||
"fmt"
|
"log"
|
||||||
"log"
|
"time"
|
||||||
"time"
|
|
||||||
|
rediscache "feedsystem_video_go/internal/middleware/redis"
|
||||||
rediscache "feedsystem_video_go/internal/middleware/redis"
|
|
||||||
|
"github.com/go-sql-driver/mysql"
|
||||||
"github.com/go-sql-driver/mysql"
|
"golang.org/x/crypto/bcrypt"
|
||||||
"golang.org/x/crypto/bcrypt"
|
"gorm.io/gorm"
|
||||||
"gorm.io/gorm"
|
)
|
||||||
)
|
|
||||||
|
type AccountService struct {
|
||||||
type AccountService struct {
|
accountRepository *AccountRepository
|
||||||
accountRepository *AccountRepository
|
cache *rediscache.Client
|
||||||
cache *rediscache.Client
|
}
|
||||||
}
|
|
||||||
|
var (
|
||||||
var (
|
ErrUsernameTaken = errors.New("username already exists")
|
||||||
ErrUsernameTaken = errors.New("username already exists")
|
ErrNewUsernameRequired = errors.New("new_username is required")
|
||||||
ErrNewUsernameRequired = errors.New("new_username is required")
|
)
|
||||||
)
|
|
||||||
|
func NewAccountService(accountRepository *AccountRepository, cache *rediscache.Client) *AccountService {
|
||||||
func NewAccountService(accountRepository *AccountRepository, cache *rediscache.Client) *AccountService {
|
return &AccountService{accountRepository: accountRepository, cache: cache}
|
||||||
return &AccountService{accountRepository: accountRepository, cache: cache}
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) CreateAccount(ctx context.Context, account *Account) error {
|
||||||
func (as *AccountService) CreateAccount(ctx context.Context, account *Account) error {
|
passwordHash, err := bcrypt.GenerateFromPassword([]byte(account.Password), bcrypt.DefaultCost)
|
||||||
passwordHash, err := bcrypt.GenerateFromPassword([]byte(account.Password), bcrypt.DefaultCost)
|
if err != nil {
|
||||||
if err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
account.Password = string(passwordHash)
|
||||||
account.Password = string(passwordHash)
|
if err := as.accountRepository.CreateAccount(ctx, account); err != nil {
|
||||||
if err := as.accountRepository.CreateAccount(ctx, account); err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
return nil
|
||||||
return nil
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) Rename(ctx context.Context, accountID uint, newUsername string) (string, error) {
|
||||||
func (as *AccountService) Rename(ctx context.Context, accountID uint, newUsername string) (string, error) {
|
if newUsername == "" {
|
||||||
if newUsername == "" {
|
return "", ErrNewUsernameRequired
|
||||||
return "", ErrNewUsernameRequired
|
}
|
||||||
}
|
|
||||||
|
token, err := auth.GenerateToken(accountID, newUsername)
|
||||||
token, err := auth.GenerateToken(accountID, newUsername)
|
if err != nil {
|
||||||
if err != nil {
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
|
||||||
|
if err := as.accountRepository.RenameWithToken(ctx, accountID, newUsername, token); err != nil {
|
||||||
if err := as.accountRepository.RenameWithToken(ctx, accountID, newUsername, token); err != nil {
|
var mysqlErr *mysql.MySQLError
|
||||||
var mysqlErr *mysql.MySQLError
|
if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
|
||||||
if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
|
return "", ErrUsernameTaken
|
||||||
return "", ErrUsernameTaken
|
}
|
||||||
}
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
if as.cache != nil {
|
||||||
if as.cache != nil {
|
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
defer cancel()
|
||||||
defer cancel()
|
|
||||||
|
if err := as.cache.SetBytes(cacheCtx, as.cache.Key("account:%d", accountID), []byte(token), 24*time.Hour); err != nil {
|
||||||
if err := as.cache.SetBytes(cacheCtx, fmt.Sprintf("account:%d", accountID), []byte(token), 24*time.Hour); err != nil {
|
log.Printf("failed to set cache: %v", err)
|
||||||
log.Printf("failed to set cache: %v", err)
|
}
|
||||||
}
|
}
|
||||||
}
|
return token, nil
|
||||||
return token, nil
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) ChangePassword(ctx context.Context, username, oldPassword, newPassword string) error {
|
||||||
func (as *AccountService) ChangePassword(ctx context.Context, username, oldPassword, newPassword string) error {
|
account, err := as.FindByUsername(ctx, username)
|
||||||
account, err := as.FindByUsername(ctx, username)
|
if err != nil {
|
||||||
if err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(oldPassword)); err != nil {
|
||||||
if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(oldPassword)); err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
passwordHash, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost)
|
||||||
passwordHash, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost)
|
if err != nil {
|
||||||
if err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
if err := as.accountRepository.ChangePassword(ctx, account.ID, string(passwordHash)); err != nil {
|
||||||
if err := as.accountRepository.ChangePassword(ctx, account.ID, string(passwordHash)); err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
if err := as.Logout(ctx, account.ID); err != nil {
|
||||||
if err := as.Logout(ctx, account.ID); err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
return nil
|
||||||
return nil
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) FindByID(ctx context.Context, id uint) (*Account, error) {
|
||||||
func (as *AccountService) FindByID(ctx context.Context, id uint) (*Account, error) {
|
if account, err := as.accountRepository.FindByID(ctx, id); err != nil {
|
||||||
if account, err := as.accountRepository.FindByID(ctx, id); err != nil {
|
return nil, err
|
||||||
return nil, err
|
} else {
|
||||||
} else {
|
return account, nil
|
||||||
return account, nil
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) FindByUsername(ctx context.Context, username string) (*Account, error) {
|
||||||
func (as *AccountService) FindByUsername(ctx context.Context, username string) (*Account, error) {
|
if account, err := as.accountRepository.FindByUsername(ctx, username); err != nil {
|
||||||
if account, err := as.accountRepository.FindByUsername(ctx, username); err != nil {
|
return nil, err
|
||||||
return nil, err
|
} else {
|
||||||
} else {
|
return account, nil
|
||||||
return account, nil
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) Login(ctx context.Context, username, password string) (string, error) {
|
||||||
func (as *AccountService) Login(ctx context.Context, username, password string) (string, error) {
|
account, err := as.FindByUsername(ctx, username)
|
||||||
account, err := as.FindByUsername(ctx, username)
|
if err != nil {
|
||||||
if err != nil {
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(password)); err != nil {
|
||||||
if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(password)); err != nil {
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
// generate token
|
||||||
// generate token
|
token, err := auth.GenerateToken(account.ID, account.Username)
|
||||||
token, err := auth.GenerateToken(account.ID, account.Username)
|
if err != nil {
|
||||||
if err != nil {
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
if err := as.accountRepository.Login(ctx, account.ID, token); err != nil {
|
||||||
if err := as.accountRepository.Login(ctx, account.ID, token); err != nil {
|
return "", err
|
||||||
return "", err
|
}
|
||||||
}
|
if as.cache != nil {
|
||||||
if as.cache != nil {
|
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
defer cancel()
|
||||||
defer cancel()
|
|
||||||
|
if err := as.cache.SetBytes(cacheCtx, as.cache.Key("account:%d", account.ID), []byte(token), 24*time.Hour); err != nil {
|
||||||
if err := as.cache.SetBytes(cacheCtx, fmt.Sprintf("account:%d", account.ID), []byte(token), 24*time.Hour); err != nil {
|
log.Printf("failed to set cache: %v", err)
|
||||||
log.Printf("failed to set cache: %v", err)
|
}
|
||||||
}
|
}
|
||||||
}
|
return token, nil
|
||||||
return token, nil
|
}
|
||||||
}
|
|
||||||
|
func (as *AccountService) Logout(ctx context.Context, accountID uint) error {
|
||||||
func (as *AccountService) Logout(ctx context.Context, accountID uint) error {
|
account, err := as.FindByID(ctx, accountID)
|
||||||
account, err := as.FindByID(ctx, accountID)
|
if err != nil {
|
||||||
if err != nil {
|
return err
|
||||||
return err
|
}
|
||||||
}
|
if account.Token == "" {
|
||||||
if account.Token == "" {
|
return nil
|
||||||
return nil
|
}
|
||||||
}
|
if as.cache != nil {
|
||||||
if as.cache != nil {
|
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
defer cancel()
|
||||||
defer cancel()
|
|
||||||
|
if err := as.cache.Del(cacheCtx, as.cache.Key("account:%d", account.ID)); err != nil {
|
||||||
if err := as.cache.Del(cacheCtx, fmt.Sprintf("account:%d", account.ID)); err != nil {
|
log.Printf("failed to del cache: %v", err)
|
||||||
log.Printf("failed to del cache: %v", err)
|
}
|
||||||
}
|
}
|
||||||
}
|
return as.accountRepository.Logout(ctx, account.ID)
|
||||||
return as.accountRepository.Logout(ctx, account.ID)
|
}
|
||||||
}
|
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi
|
|||||||
//L1:本地缓存
|
//L1:本地缓存
|
||||||
var missedL1 []uint
|
var missedL1 []uint
|
||||||
for _, id := range videoIDs {
|
for _, id := range videoIDs {
|
||||||
cacheKey := fmt.Sprintf("video:entity:%d", id)
|
cacheKey := f.rediscache.Key("video:entity:%d", id)
|
||||||
if f.localcache != nil {
|
if f.localcache != nil {
|
||||||
if v, found := f.localcache.Get(cacheKey); found {
|
if v, found := f.localcache.Get(cacheKey); found {
|
||||||
if data, ok := v.(video.Video); ok {
|
if data, ok := v.(video.Video); ok {
|
||||||
@@ -66,7 +66,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi
|
|||||||
if len(missedL1) > 0 {
|
if len(missedL1) > 0 {
|
||||||
cacheKeys := make([]string, len(missedL1))
|
cacheKeys := make([]string, len(missedL1))
|
||||||
for i, id := range missedL1 {
|
for i, id := range missedL1 {
|
||||||
cacheKeys[i] = fmt.Sprintf("video:entity:%d", id)
|
cacheKeys[i] = f.rediscache.Key("video:entity:%d", id)
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
@@ -109,7 +109,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi
|
|||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func(videoID uint) {
|
go func(videoID uint) {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
sfKey := fmt.Sprintf("sf:entity:%d", videoID)
|
sfKey := f.rediscache.Key("sf:entity:%d", videoID)
|
||||||
|
|
||||||
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
||||||
videoList, err := f.repo.GetByIDs(ctx, []uint{videoID})
|
videoList, err := f.repo.GetByIDs(ctx, []uint{videoID})
|
||||||
@@ -119,7 +119,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi
|
|||||||
}
|
}
|
||||||
|
|
||||||
safeCopy := *videoList[0]
|
safeCopy := *videoList[0]
|
||||||
cachekey := fmt.Sprintf("video:entity:%d", safeCopy.ID)
|
cachekey := f.rediscache.Key("video:entity:%d", safeCopy.ID)
|
||||||
if b, err := json.Marshal(safeCopy); err == nil {
|
if b, err := json.Marshal(safeCopy); err == nil {
|
||||||
//异步回写redis
|
//异步回写redis
|
||||||
go func(k string, b []byte) {
|
go func(k string, b []byte) {
|
||||||
@@ -137,7 +137,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi
|
|||||||
mu.Lock()
|
mu.Lock()
|
||||||
videoMap[id] = &safeCopy
|
videoMap[id] = &safeCopy
|
||||||
mu.Unlock()
|
mu.Unlock()
|
||||||
f.localcache.Set(fmt.Sprintf("video:entity:%d", safeCopy.ID), safeCopy, 5*time.Second)
|
f.localcache.Set(f.rediscache.Key("video:entity:%d", safeCopy.ID), safeCopy, 5*time.Second)
|
||||||
}
|
}
|
||||||
}(id)
|
}(id)
|
||||||
}
|
}
|
||||||
@@ -148,7 +148,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi
|
|||||||
// 查询最新视频 (冷热分离 + 游标分页)
|
// 查询最新视频 (冷热分离 + 游标分页)
|
||||||
func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) {
|
func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) {
|
||||||
// 获取 ZSET 中最老的一条数据
|
// 获取 ZSET 中最老的一条数据
|
||||||
zsetTail, err := f.rediscache.ZRangeWithScores(ctx, "feed:global_timeline", 0, 0)
|
zsetTail, err := f.rediscache.ZRangeWithScores(ctx, f.rediscache.Key("feed:global_timeline"), 0, 0)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ListLatestResponse{}, err
|
return ListLatestResponse{}, err
|
||||||
@@ -158,7 +158,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti
|
|||||||
|
|
||||||
if isZsetEmpty {
|
if isZsetEmpty {
|
||||||
//全局静态锁：无视所有用户的不同时间戳游标
|
//全局静态锁：无视所有用户的不同时间戳游标
|
||||||
sfKey := "sf:fallback:global_timeline_rebuild"
|
sfKey := f.rediscache.Key("sf:fallback:global_timeline_rebuild")
|
||||||
|
|
||||||
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
||||||
// 无视游标，直接去 MySQL 捞最新的 1000 条
|
// 无视游标，直接去 MySQL 捞最新的 1000 条
|
||||||
@@ -180,7 +180,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti
|
|||||||
Member: fmt.Sprintf("%d", vid.ID),
|
Member: fmt.Sprintf("%d", vid.ID),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
f.rediscache.ZAdd(bgCtx, "feed:global_timeline", zElements...)
|
f.rediscache.ZAdd(bgCtx, f.rediscache.Key("feed:global_timeline"), zElements...)
|
||||||
return "SUCCESS", nil
|
return "SUCCESS", nil
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -207,7 +207,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti
|
|||||||
//冷数据降级查库
|
//冷数据降级查库
|
||||||
|
|
||||||
// 针对个别用户的防并发（此时可以用时间戳做锁，因为冷尾流量极小）
|
// 针对个别用户的防并发（此时可以用时间戳做锁，因为冷尾流量极小）
|
||||||
sfKey := fmt.Sprintf("sf:cold:listLatest:%d:%d", limit, reqTime)
|
sfKey := f.rediscache.Key("sf:cold:listLatest:%d:%d", limit, reqTime)
|
||||||
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
||||||
return f.repo.ListLatest(ctx, limit, latestBefore)
|
return f.repo.ListLatest(ctx, limit, latestBefore)
|
||||||
})
|
})
|
||||||
@@ -224,7 +224,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti
|
|||||||
maxScore = fmt.Sprintf("%d", reqTime-1) // 防重复
|
maxScore = fmt.Sprintf("%d", reqTime-1) // 防重复
|
||||||
}
|
}
|
||||||
|
|
||||||
videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, "feed:global_timeline", maxScore, "-inf", 0, int64(limit))
|
videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, f.rediscache.Key("feed:global_timeline"), maxScore, "-inf", 0, int64(limit))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ListLatestResponse{}, err
|
return ListLatestResponse{}, err
|
||||||
}
|
}
|
||||||
@@ -254,7 +254,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti
|
|||||||
coldCursor = latestBefore
|
coldCursor = latestBefore
|
||||||
}
|
}
|
||||||
|
|
||||||
sfKey := fmt.Sprintf("sf:stitch:listLatest:%d:%d", remainLimit, coldCursor.UnixMilli())
|
sfKey := f.rediscache.Key("sf:stitch:listLatest:%d:%d", remainLimit, coldCursor.UnixMilli())
|
||||||
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) {
|
||||||
return f.repo.ListLatest(ctx, remainLimit, coldCursor)
|
return f.repo.ListLatest(ctx, remainLimit, coldCursor)
|
||||||
})
|
})
|
||||||
@@ -343,7 +343,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo
|
|||||||
if !latestBefore.IsZero() {
|
if !latestBefore.IsZero() {
|
||||||
before = latestBefore.Unix()
|
before = latestBefore.Unix()
|
||||||
}
|
}
|
||||||
cacheKey = fmt.Sprintf("feed:listByFollowing:limit=%d:accountID=%d:before=%d", limit, viewerAccountID, before)
|
cacheKey = f.rediscache.Key("feed:listByFollowing:limit=%d:accountID=%d:before=%d", limit, viewerAccountID, before)
|
||||||
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
@@ -413,10 +413,10 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i
|
|||||||
const win = 60
|
const win = 60
|
||||||
keys := make([]string, 0, win)
|
keys := make([]string, 0, win)
|
||||||
for i := 0; i < win; i++ {
|
for i := 0; i < win; i++ {
|
||||||
keys = append(keys, "hot:video:1m:"+asOf.Add(-time.Duration(i)*time.Minute).Format("200601021504"))
|
keys = append(keys, f.rediscache.Key("hot:video:1m:%s", asOf.Add(-time.Duration(i)*time.Minute).Format("200601021504")))
|
||||||
}
|
}
|
||||||
|
|
||||||
dest := "hot:video:merge:1m:" + asOf.Format("200601021504") // 快照key：同一个as_of页内复用
|
dest := f.rediscache.Key("hot:video:merge:1m:%s", asOf.Format("200601021504")) // 快照key：同一个as_of页内复用
|
||||||
opCtx, cancel := context.WithTimeout(ctx, 80*time.Millisecond)
|
opCtx, cancel := context.WithTimeout(ctx, 80*time.Millisecond)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
|
|||||||
@@ -1,140 +1,139 @@
|
|||||||
package jwt
|
package jwt
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"log"
|
||||||
"log"
|
"net/http"
|
||||||
"net/http"
|
"strings"
|
||||||
"strings"
|
"time"
|
||||||
"time"
|
|
||||||
|
"feedsystem_video_go/internal/account"
|
||||||
"feedsystem_video_go/internal/account"
|
"feedsystem_video_go/internal/auth"
|
||||||
"feedsystem_video_go/internal/auth"
|
rediscache "feedsystem_video_go/internal/middleware/redis"
|
||||||
rediscache "feedsystem_video_go/internal/middleware/redis"
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
"github.com/gin-gonic/gin"
|
)
|
||||||
)
|
|
||||||
|
// JWTAuth check jwt token and ensure it matches the currently stored token.
|
||||||
// JWTAuth check jwt token and ensure it matches the currently stored token.
|
func JWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc {
|
||||||
func JWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc {
|
return func(c *gin.Context) {
|
||||||
return func(c *gin.Context) {
|
authHeader := c.GetHeader("Authorization")
|
||||||
authHeader := c.GetHeader("Authorization")
|
if authHeader == "" {
|
||||||
if authHeader == "" {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing authorization header"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing authorization header"})
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
parts := strings.SplitN(authHeader, " ", 2)
|
||||||
parts := strings.SplitN(authHeader, " ", 2)
|
if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") {
|
||||||
if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"})
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
tokenString := parts[1]
|
||||||
tokenString := parts[1]
|
|
||||||
|
claims, err := auth.ParseToken(tokenString)
|
||||||
claims, err := auth.ParseToken(tokenString)
|
if err != nil {
|
||||||
if err != nil {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"})
|
return
|
||||||
return
|
}
|
||||||
}
|
check(c, claims, tokenString, accountRepo, cache)
|
||||||
check(c, claims, tokenString, accountRepo, cache)
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
func SoftJWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc {
|
||||||
func SoftJWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc {
|
return func(c *gin.Context) {
|
||||||
return func(c *gin.Context) {
|
authHeader := c.GetHeader("Authorization")
|
||||||
authHeader := c.GetHeader("Authorization")
|
if authHeader == "" {
|
||||||
if authHeader == "" {
|
c.Next()
|
||||||
c.Next()
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
parts := strings.SplitN(authHeader, " ", 2)
|
||||||
parts := strings.SplitN(authHeader, " ", 2)
|
if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") {
|
||||||
if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"})
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
tokenString := parts[1]
|
||||||
tokenString := parts[1]
|
|
||||||
|
claims, err := auth.ParseToken(tokenString)
|
||||||
claims, err := auth.ParseToken(tokenString)
|
if err != nil {
|
||||||
if err != nil {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"})
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
check(c, claims, tokenString, accountRepo, cache)
|
||||||
check(c, claims, tokenString, accountRepo, cache)
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
func check(c *gin.Context, claims *auth.Claims, tokenString string, accountRepo *account.AccountRepository, cache *rediscache.Client) {
|
||||||
func check(c *gin.Context, claims *auth.Claims, tokenString string, accountRepo *account.AccountRepository, cache *rediscache.Client) {
|
key := cache.Key("account:%d", claims.AccountID)
|
||||||
key := fmt.Sprintf("account:%d", claims.AccountID)
|
|
||||||
|
// 先查 Redis
|
||||||
// 先查 Redis
|
if cache != nil {
|
||||||
if cache != nil {
|
cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond)
|
||||||
cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond)
|
defer cancel()
|
||||||
defer cancel()
|
|
||||||
|
b, err := cache.GetBytes(cacheCtx, key)
|
||||||
b, err := cache.GetBytes(cacheCtx, key)
|
if err == nil {
|
||||||
if err == nil {
|
if string(b) != tokenString {
|
||||||
if string(b) != tokenString {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"})
|
return
|
||||||
return
|
}
|
||||||
}
|
c.Set("accountID", claims.AccountID)
|
||||||
c.Set("accountID", claims.AccountID)
|
c.Set("username", claims.Username)
|
||||||
c.Set("username", claims.Username)
|
c.Next()
|
||||||
c.Next()
|
return
|
||||||
return
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
// Redis 故障/未启用：查 DB 兜底
|
||||||
// Redis 故障/未启用：查 DB 兜底
|
accountInfo, err := accountRepo.FindByID(c.Request.Context(), claims.AccountID)
|
||||||
accountInfo, err := accountRepo.FindByID(c.Request.Context(), claims.AccountID)
|
if err != nil || accountInfo.Token == "" || accountInfo.Token != tokenString {
|
||||||
if err != nil || accountInfo.Token == "" || accountInfo.Token != tokenString {
|
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"})
|
||||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"})
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
if cache != nil {
|
||||||
if cache != nil {
|
cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond)
|
||||||
cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond)
|
defer cancel()
|
||||||
defer cancel()
|
|
||||||
|
if err := cache.SetBytes(cacheCtx, key, []byte(tokenString), 24*time.Hour); err != nil {
|
||||||
if err := cache.SetBytes(cacheCtx, key, []byte(tokenString), 24*time.Hour); err != nil {
|
log.Printf("failed to set cache: %v", err)
|
||||||
log.Printf("failed to set cache: %v", err)
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
c.Set("accountID", claims.AccountID)
|
||||||
c.Set("accountID", claims.AccountID)
|
c.Set("username", claims.Username)
|
||||||
c.Set("username", claims.Username)
|
c.Next()
|
||||||
c.Next()
|
|
||||||
|
}
|
||||||
}
|
|
||||||
|
func GetAccountID(c *gin.Context) (uint, error) {
|
||||||
func GetAccountID(c *gin.Context) (uint, error) {
|
uidValue, exists := c.Get("accountID")
|
||||||
uidValue, exists := c.Get("accountID")
|
if !exists {
|
||||||
if !exists {
|
return 0, errors.New("accountID not found")
|
||||||
return 0, errors.New("accountID not found")
|
}
|
||||||
}
|
|
||||||
|
accountID, ok := uidValue.(uint)
|
||||||
accountID, ok := uidValue.(uint)
|
if !ok {
|
||||||
if !ok {
|
return 0, errors.New("accountID has invalid type")
|
||||||
return 0, errors.New("accountID has invalid type")
|
}
|
||||||
}
|
|
||||||
|
return accountID, nil
|
||||||
return accountID, nil
|
}
|
||||||
}
|
|
||||||
|
func GetUsername(c *gin.Context) (string, error) {
|
||||||
func GetUsername(c *gin.Context) (string, error) {
|
val, exists := c.Get("username")
|
||||||
val, exists := c.Get("username")
|
if !exists {
|
||||||
if !exists {
|
return "", errors.New("username not found")
|
||||||
return "", errors.New("username not found")
|
}
|
||||||
}
|
|
||||||
|
username, ok := val.(string)
|
||||||
username, ok := val.(string)
|
if !ok {
|
||||||
if !ok {
|
return "", errors.New("username has invalid type")
|
||||||
return "", errors.New("username has invalid type")
|
}
|
||||||
}
|
|
||||||
|
return username, nil
|
||||||
return username, nil
|
}
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,99 +1,111 @@
|
|||||||
package redis
|
package redis
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"feedsystem_video_go/internal/config"
|
"feedsystem_video_go/internal/config"
|
||||||
"strconv"
|
"fmt"
|
||||||
"time"
|
"strconv"
|
||||||
|
"time"
|
||||||
redis "github.com/redis/go-redis/v9"
|
|
||||||
)
|
redis "github.com/redis/go-redis/v9"
|
||||||
|
)
|
||||||
type Client struct {
|
|
||||||
rdb *redis.Client
|
type Client struct {
|
||||||
}
|
rdb *redis.Client
|
||||||
|
keyPrefix string
|
||||||
func NewFromEnv(cfg *config.RedisConfig) (*Client, error) {
|
}
|
||||||
rdb := redis.NewClient(&redis.Options{
|
|
||||||
Addr: cfg.Host + ":" + strconv.Itoa(cfg.Port),
|
const defaultKeyPrefix = "v1:"
|
||||||
Password: cfg.Password,
|
|
||||||
DB: cfg.DB,
|
func NewFromEnv(cfg *config.RedisConfig) (*Client, error) {
|
||||||
})
|
rdb := redis.NewClient(&redis.Options{
|
||||||
return &Client{rdb: rdb}, nil
|
Addr: cfg.Host + ":" + strconv.Itoa(cfg.Port),
|
||||||
}
|
Password: cfg.Password,
|
||||||
|
DB: cfg.DB,
|
||||||
func (c *Client) Close() error {
|
})
|
||||||
if c == nil || c.rdb == nil {
|
return &Client{rdb: rdb, keyPrefix: defaultKeyPrefix}, nil
|
||||||
return nil
|
}
|
||||||
}
|
|
||||||
return c.rdb.Close()
|
func (c *Client) Close() error {
|
||||||
}
|
if c == nil || c.rdb == nil {
|
||||||
|
return nil
|
||||||
func (c *Client) Ping(ctx context.Context) error {
|
}
|
||||||
if c == nil || c.rdb == nil {
|
return c.rdb.Close()
|
||||||
return nil
|
}
|
||||||
}
|
|
||||||
return c.rdb.Ping(ctx).Err()
|
func (c *Client) Ping(ctx context.Context) error {
|
||||||
}
|
if c == nil || c.rdb == nil {
|
||||||
|
return nil
|
||||||
func IsMiss(err error) bool {
|
}
|
||||||
return err == redis.Nil
|
return c.rdb.Ping(ctx).Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
func randToken(n int) (string, error) {
|
func IsMiss(err error) bool {
|
||||||
b := make([]byte, n)
|
return err == redis.Nil
|
||||||
if _, err := rand.Read(b); err != nil {
|
}
|
||||||
return "", err
|
|
||||||
}
|
func (c *Client) Key(format string, args ...any) string {
|
||||||
return hex.EncodeToString(b), nil
|
prefix := ""
|
||||||
}
|
if c != nil {
|
||||||
|
prefix = c.keyPrefix
|
||||||
func (c *Client) Lock(ctx context.Context, key string, ttl time.Duration) (token string, ok bool, err error) {
|
}
|
||||||
if c == nil || c.rdb == nil {
|
return prefix + fmt.Sprintf(format, args...)
|
||||||
return "", false, nil
|
}
|
||||||
}
|
|
||||||
token, err = randToken(16)
|
func randToken(n int) (string, error) {
|
||||||
if err != nil {
|
b := make([]byte, n)
|
||||||
return "", false, err
|
if _, err := rand.Read(b); err != nil {
|
||||||
}
|
return "", err
|
||||||
ok, err = c.rdb.SetNX(ctx, key, token, ttl).Result()
|
}
|
||||||
return token, ok, err
|
return hex.EncodeToString(b), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
var unlockScript = redis.NewScript(`
|
func (c *Client) Lock(ctx context.Context, key string, ttl time.Duration) (token string, ok bool, err error) {
|
||||||
if redis.call("GET", KEYS[1]) == ARGV[1] then
|
if c == nil || c.rdb == nil {
|
||||||
return redis.call("DEL", KEYS[1])
|
return "", false, nil
|
||||||
else
|
}
|
||||||
return 0
|
token, err = randToken(16)
|
||||||
end
|
if err != nil {
|
||||||
`)
|
return "", false, err
|
||||||
|
}
|
||||||
var incrementWithExpireScript = redis.NewScript(`
|
ok, err = c.rdb.SetNX(ctx, key, token, ttl).Result()
|
||||||
local count = redis.call("INCR", KEYS[1])
|
return token, ok, err
|
||||||
if count == 1 then
|
}
|
||||||
redis.call("PEXPIRE", KEYS[1], ARGV[1])
|
|
||||||
end
|
var unlockScript = redis.NewScript(`
|
||||||
return count
|
if redis.call("GET", KEYS[1]) == ARGV[1] then
|
||||||
`)
|
return redis.call("DEL", KEYS[1])
|
||||||
|
else
|
||||||
func (c *Client) Unlock(ctx context.Context, key string, token string) error {
|
return 0
|
||||||
if c == nil || c.rdb == nil {
|
end
|
||||||
return nil
|
`)
|
||||||
}
|
|
||||||
_, err := unlockScript.Run(ctx, c.rdb, []string{key}, token).Result()
|
var incrementWithExpireScript = redis.NewScript(`
|
||||||
return err
|
local count = redis.call("INCR", KEYS[1])
|
||||||
}
|
if count == 1 then
|
||||||
|
redis.call("PEXPIRE", KEYS[1], ARGV[1])
|
||||||
func (c *Client) IncrementWithExpire(ctx context.Context, key string, expire time.Duration) (int64, error) {
|
end
|
||||||
if c == nil || c.rdb == nil {
|
return count
|
||||||
return 0, nil
|
`)
|
||||||
}
|
|
||||||
return incrementWithExpireScript.Run(
|
func (c *Client) Unlock(ctx context.Context, key string, token string) error {
|
||||||
ctx,
|
if c == nil || c.rdb == nil {
|
||||||
c.rdb,
|
return nil
|
||||||
[]string{key},
|
}
|
||||||
expire.Milliseconds(),
|
_, err := unlockScript.Run(ctx, c.rdb, []string{key}, token).Result()
|
||||||
).Int64()
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (c *Client) IncrementWithExpire(ctx context.Context, key string, expire time.Duration) (int64, error) {
|
||||||
|
if c == nil || c.rdb == nil {
|
||||||
|
return 0, nil
|
||||||
|
}
|
||||||
|
return incrementWithExpireScript.Run(
|
||||||
|
ctx,
|
||||||
|
c.rdb,
|
||||||
|
[]string{key},
|
||||||
|
expire.Milliseconds(),
|
||||||
|
).Int64()
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,29 +1,28 @@
|
|||||||
package video
|
package video
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"strconv"
|
||||||
"strconv"
|
"time"
|
||||||
"time"
|
|
||||||
|
rediscache "feedsystem_video_go/internal/middleware/redis"
|
||||||
rediscache "feedsystem_video_go/internal/middleware/redis"
|
)
|
||||||
)
|
|
||||||
|
// 更新视频流行度缓存
|
||||||
// 更新视频流行度缓存
|
func UpdatePopularityCache(ctx context.Context, cache *rediscache.Client, id uint, change int64) {
|
||||||
func UpdatePopularityCache(ctx context.Context, cache *rediscache.Client, id uint, change int64) {
|
if cache == nil || id == 0 || change == 0 {
|
||||||
if cache == nil || id == 0 || change == 0 {
|
return
|
||||||
return
|
}
|
||||||
}
|
|
||||||
|
_ = cache.Del(context.Background(), cache.Key("video:detail:id=%d", id))
|
||||||
_ = cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id))
|
|
||||||
|
now := time.Now().UTC().Truncate(time.Minute)
|
||||||
now := time.Now().UTC().Truncate(time.Minute)
|
windowKey := cache.Key("hot:video:1m:%s", now.Format("200601021504"))
|
||||||
windowKey := "hot:video:1m:" + now.Format("200601021504")
|
member := strconv.FormatUint(uint64(id), 10)
|
||||||
member := strconv.FormatUint(uint64(id), 10)
|
|
||||||
|
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
defer cancel()
|
||||||
defer cancel()
|
|
||||||
|
_ = cache.ZincrBy(opCtx, windowKey, member, float64(change))
|
||||||
_ = cache.ZincrBy(opCtx, windowKey, member, float64(change))
|
_ = cache.Expire(opCtx, windowKey, 2*time.Hour)
|
||||||
_ = cache.Expire(opCtx, windowKey, 2*time.Hour)
|
}
|
||||||
}
|
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
@@ -83,7 +82,7 @@ func (vs *VideoService) Delete(ctx context.Context, id uint, authorID uint) erro
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if vs.cache != nil {
|
if vs.cache != nil {
|
||||||
cacheKey := fmt.Sprintf("video:detail:id=%d", id)
|
cacheKey := vs.cache.Key("video:detail:id=%d", id)
|
||||||
_ = vs.cache.Del(context.Background(), cacheKey)
|
_ = vs.cache.Del(context.Background(), cacheKey)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -98,7 +97,7 @@ func (vs *VideoService) ListByAuthorID(ctx context.Context, authorID uint) ([]Vi
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (vs *VideoService) GetDetail(ctx context.Context, id uint) (*Video, error) {
|
func (vs *VideoService) GetDetail(ctx context.Context, id uint) (*Video, error) {
|
||||||
cacheKey := fmt.Sprintf("video:detail:id=%d", id)
|
cacheKey := vs.cache.Key("video:detail:id=%d", id)
|
||||||
|
|
||||||
getCached := func() (*Video, bool) {
|
getCached := func() (*Video, bool) {
|
||||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
@@ -204,11 +203,11 @@ func (vs *VideoService) UpdatePopularity(ctx context.Context, id uint, change in
|
|||||||
|
|
||||||
if vs.cache != nil {
|
if vs.cache != nil {
|
||||||
// 1) 详情缓存：直接失效（最简单靠谱）
|
// 1) 详情缓存：直接失效（最简单靠谱）
|
||||||
_ = vs.cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id))
|
_ = vs.cache.Del(context.Background(), vs.cache.Key("video:detail:id=%d", id))
|
||||||
|
|
||||||
// 2) 热榜：写到“时间窗ZSET”，不要用 detail key
|
// 2) 热榜：写到“时间窗ZSET”，不要用 detail key
|
||||||
now := time.Now().UTC().Truncate(time.Minute)
|
now := time.Now().UTC().Truncate(time.Minute)
|
||||||
windowKey := "hot:video:1m:" + now.Format("200601021504")
|
windowKey := vs.cache.Key("hot:video:1m:%s", now.Format("200601021504"))
|
||||||
member := strconv.FormatUint(uint64(id), 10)
|
member := strconv.FormatUint(uint64(id), 10)
|
||||||
|
|
||||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||||
|
|||||||
@@ -1,93 +1,93 @@
|
|||||||
package worker
|
package worker
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"feedsystem_video_go/internal/middleware/rabbitmq"
|
"feedsystem_video_go/internal/middleware/rabbitmq"
|
||||||
"feedsystem_video_go/internal/middleware/redis"
|
"feedsystem_video_go/internal/middleware/redis"
|
||||||
"feedsystem_video_go/internal/video"
|
"feedsystem_video_go/internal/video"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
oredis "github.com/redis/go-redis/v9"
|
oredis "github.com/redis/go-redis/v9"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
)
|
)
|
||||||
|
|
||||||
func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) {
|
func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) {
|
||||||
go func() {
|
go func() {
|
||||||
for {
|
for {
|
||||||
var messages []video.OutboxMsg
|
var messages []video.OutboxMsg
|
||||||
|
|
||||||
err := db.Where("status = ?", "pending").Order("create_time ASC").Limit(100).Find(&messages).Error
|
err := db.Where("status = ?", "pending").Order("create_time ASC").Limit(100).Find(&messages).Error
|
||||||
|
|
||||||
if err != nil || len(messages) == 0 {
|
if err != nil || len(messages) == 0 {
|
||||||
time.Sleep(1 * time.Second)
|
time.Sleep(1 * time.Second)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, msg := range messages {
|
for _, msg := range messages {
|
||||||
err := tmq.PublishVideo(context.Background(), msg.VideoID, msg.CreateTime)
|
err := tmq.PublishVideo(context.Background(), msg.VideoID, msg.CreateTime)
|
||||||
|
|
||||||
if err == nil {
|
if err == nil {
|
||||||
db.Delete(&msg)
|
db.Delete(&msg)
|
||||||
} else {
|
} else {
|
||||||
log.Printf("投递MQ失败: VideoID: %d, err: %v", msg.VideoID, err)
|
log.Printf("投递MQ失败: VideoID: %d, err: %v", msg.VideoID, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redis.Client) {
|
func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redis.Client) {
|
||||||
msgs, err := tmq.Ch.Consume(
|
msgs, err := tmq.Ch.Consume(
|
||||||
queueName,
|
queueName,
|
||||||
"",
|
"",
|
||||||
false,
|
false,
|
||||||
false,
|
false,
|
||||||
false,
|
false,
|
||||||
false,
|
false,
|
||||||
nil,
|
nil,
|
||||||
)
|
)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("注册消费失败")
|
log.Printf("注册消费失败")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
for msg := range msgs {
|
for msg := range msgs {
|
||||||
var event rabbitmq.TimelineEvent
|
var event rabbitmq.TimelineEvent
|
||||||
err := json.Unmarshal(msg.Body, &event)
|
err := json.Unmarshal(msg.Body, &event)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("反序列化失败")
|
log.Printf("反序列化失败")
|
||||||
msg.Ack(false)
|
msg.Ack(false)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||||
timelineKey := "feed:global_timeline"
|
timelineKey := redisClient.Key("feed:global_timeline")
|
||||||
err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{
|
err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{
|
||||||
Score: float64(event.CreateTime),
|
Score: float64(event.CreateTime),
|
||||||
Member: fmt.Sprintf("%d", event.VideoID),
|
Member: fmt.Sprintf("%d", event.VideoID),
|
||||||
})
|
})
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("写入Zset失败")
|
log.Printf("写入Zset失败")
|
||||||
msg.Nack(false, true)
|
msg.Nack(false, true)
|
||||||
cancel()
|
cancel()
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
err = redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001)
|
err = redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("ZRem失败")
|
log.Printf("ZRem失败")
|
||||||
}
|
}
|
||||||
|
|
||||||
msg.Ack(false)
|
msg.Ack(false)
|
||||||
cancel()
|
cancel()
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user