Compare commits

...

3 Commits

Author SHA1 Message Date
137b2a9738 var cache increaseUpdate 2023-11-29 18:24:41 +08:00
9f49a274cd fix bugs 2023-11-29 00:00:25 +08:00
7c3f8baaa2 map cache add increase interface 2023-11-28 22:46:22 +08:00
14 changed files with 348 additions and 130 deletions

View File

@ -6,6 +6,7 @@ import (
"github.com/fthvgb1/wp-go/app/pkg/dao"
"github.com/fthvgb1/wp-go/app/pkg/logs"
"github.com/fthvgb1/wp-go/app/pkg/models"
"github.com/fthvgb1/wp-go/cache"
"github.com/fthvgb1/wp-go/cache/cachemanager"
"github.com/fthvgb1/wp-go/helper/slice"
"github.com/fthvgb1/wp-go/safety"
@ -51,9 +52,15 @@ func InitActionsCommonCache() {
return config.GetConfig().CacheTime.RecentCommentsCacheTime
})
cachemanager.NewMemoryMapCache(nil, dao.PostComments, c.CacheTime.PostCommentsCacheTime, "postCommentIds", func() time.Duration {
cachemanager.NewMemoryMapCache(nil, dao.PostComments, c.CacheTime.PostCommentsCacheTime,
"postCommentIds",
func() time.Duration {
return config.GetConfig().CacheTime.PostCommentsCacheTime
})
},
cache.NewIncreaseUpdate("comment-increaseUpdate", dao.GetIncreaseComment, 30*time.Second,
func() time.Duration {
return config.GetConfig().CacheTime.CommentsIncreaseUpdateTime
}))
cachemanager.NewVarMemoryCache(dao.GetMaxPostId, c.CacheTime.MaxPostIdCacheTime, "maxPostId", func() time.Duration {
return config.GetConfig().CacheTime.MaxPostIdCacheTime

View File

@ -63,6 +63,7 @@ type CacheTime struct {
UserInfoCacheTime time.Duration `yaml:"userInfoCacheTime" json:"userInfoCacheTime,omitempty"`
CommentsCacheTime time.Duration `yaml:"commentsCacheTime" json:"commentsCacheTime,omitempty"`
SleepTime []time.Duration `yaml:"sleepTime" json:"sleepTime,omitempty"`
CommentsIncreaseUpdateTime time.Duration `yaml:"commentsIncreaseUpdateTime" json:"commentsIncreaseUpdateTime"`
}
type Ssl struct {

View File

@ -2,11 +2,14 @@ package dao
import (
"context"
"database/sql"
"errors"
"github.com/fthvgb1/wp-go/app/pkg/models"
"github.com/fthvgb1/wp-go/helper"
"github.com/fthvgb1/wp-go/helper/number"
"github.com/fthvgb1/wp-go/helper/slice"
"github.com/fthvgb1/wp-go/model"
"time"
)
// RecentComments
@ -49,14 +52,63 @@ func PostComments(ctx context.Context, postId uint64, _ ...any) ([]uint64, error
}
func GetCommentByIds(ctx context.Context, ids []uint64, _ ...any) (map[uint64]models.Comments, error) {
if len(ids) < 1 {
return nil, nil
}
m := make(map[uint64]models.Comments)
r, err := model.SimpleFind[models.Comments](ctx, model.SqlBuilder{
off := 0
for {
id := slice.Slice(ids, off, 500)
if len(id) < 1 {
break
}
r, err := model.Finds[models.Comments](ctx, model.Conditions(
model.Where(model.SqlBuilder{
{"comment_ID", "in", ""}, {"comment_approved", "1"},
}, "*", slice.ToAnySlice(ids))
}),
model.Fields("*"),
model.In(slice.ToAnySlice(id)),
))
if err != nil {
return m, err
}
return slice.SimpleToMap(r, func(t models.Comments) uint64 {
return t.CommentId
}), err
for _, comments := range r {
m[comments.CommentId] = comments
}
off += 500
}
return m, nil
}
func GetIncreaseComment(ctx context.Context, currentData []uint64, k uint64, t time.Time, _ ...any) (data []uint64, save bool, refresh bool, err error) {
r, err := model.ChunkFind[models.Comments](ctx, 1000, model.Conditions(
model.Where(model.SqlBuilder{
{"comment_approved", "1"},
{"comment_post_ID", "=", number.IntToString(k), "int"},
{"comment_date", ">=", t.Format(time.DateTime)},
}),
model.Fields("comment_ID"),
model.Order(model.SqlBuilder{
{"comment_date_gmt", "asc"},
{"comment_ID", "asc"},
})),
)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
err = nil
refresh = true
}
return
}
if len(r) < 1 {
refresh = true
return
}
rr := slice.Map(r, func(t models.Comments) uint64 {
return t.CommentId
})
data = append(currentData, rr...)
save = true
return
}

7
cache/cache.go vendored
View File

@ -30,3 +30,10 @@ type AnyCache[T any] interface {
Flush(ctx context.Context)
GetLastSetTime(ctx context.Context) time.Time
}
type Refresh[K comparable, V any] interface {
Refresh(ctx context.Context, k K, a ...any)
}
type RefreshVar[T any] interface {
Refresh(ctx context.Context, a ...any)
}

View File

@ -5,6 +5,7 @@ import (
"errors"
"github.com/fthvgb1/wp-go/cache"
"github.com/fthvgb1/wp-go/cache/reload"
"github.com/fthvgb1/wp-go/helper"
str "github.com/fthvgb1/wp-go/helper/strings"
"github.com/fthvgb1/wp-go/safety"
"time"
@ -15,8 +16,6 @@ var ctx = context.Background()
var mapFlush = safety.NewMap[string, func(any)]()
var anyFlush = safety.NewMap[string, func()]()
var expiredTime = safety.NewMap[string, expire]()
var varCache = safety.NewMap[string, any]()
var mapCache = safety.NewMap[string, any]()
@ -48,12 +47,6 @@ func SetMapCache[K comparable, V any](name string, ca *cache.MapCache[K, V]) err
return nil
}
type expire struct {
fn func() time.Duration
p *safety.Var[time.Duration]
isUseManger *safety.Var[bool]
}
type flush interface {
Flush(ctx context.Context)
}
@ -176,7 +169,8 @@ func parseArgs(args ...any) (string, func() time.Duration) {
}
func NewMapCache[K comparable, V any](data cache.Cache[K, V], batchFn cache.MapBatchFn[K, V], fn cache.MapSingleFn[K, V], args ...any) *cache.MapCache[K, V] {
m := cache.NewMapCache[K, V](data, fn, batchFn)
inc := helper.ParseArgs(cache.IncreaseUpdate[K, V]{}, args...)
m := cache.NewMapCache[K, V](data, fn, batchFn, inc)
FlushPush(m)
ClearPush(m)
name, f := parseArgs(args...)
@ -201,43 +195,13 @@ func SetExpireTime(c cache.SetTime, name string, expireTime time.Duration, expir
if name == "" {
return
}
var t, tt func() time.Duration
t = expireTimeFn
if t == nil {
t = func() time.Duration {
return expireTime
}
}
tt = t
expireTime = t()
p := safety.NewVar(expireTime)
e := expire{
fn: t,
p: p,
isUseManger: safety.NewVar(false),
}
expiredTime.Store(name, e)
reload.Push(func() {
if !e.isUseManger.Load() {
e.p.Store(tt())
}
}, str.Join("cacheManger-", name, "-expiredTime"))
t = func() time.Duration {
return e.p.Load()
}
c.SetExpiredTime(t)
fn := reload.FnVal(str.Join("cacheManger-", name, "-expiredTime"), expireTime, expireTimeFn)
c.SetExpiredTime(fn)
}
func ChangeExpireTime(t time.Duration, name ...string) {
for _, s := range name {
v, ok := expiredTime.Load(s)
if !ok {
continue
}
v.p.Store(t)
if !v.isUseManger.Load() {
v.isUseManger.Store(true)
}
reload.ChangeFnVal(s, t)
}
}
@ -255,7 +219,9 @@ func ClearExpired() {
}
func NewVarCache[T any](c cache.AnyCache[T], fn func(context.Context, ...any) (T, error), a ...any) *cache.VarCache[T] {
v := cache.NewVarCache(c, fn)
inc := helper.ParseArgs(cache.IncreaseUpdateVar[T]{}, a...)
ref := helper.ParseArgs(cache.RefreshVar[T](nil), a...)
v := cache.NewVarCache(c, fn, inc, ref)
FlushPush(v)
name, _ := parseArgs(a...)
if name != "" {

View File

@ -94,18 +94,18 @@ func TestSetExpireTime(t *testing.T) {
func TestSetMapCache(t *testing.T) {
t.Run("t1", func(t *testing.T) {
NewMemoryMapCache(nil, func(ctx2 context.Context, k string, a ...any) (string, error) {
x := NewMemoryMapCache(nil, func(ctx2 context.Context, k string, a ...any) (string, error) {
fmt.Println("memory cache")
return strings.Repeat(k, 2), nil
}, time.Hour, "test")
fmt.Println(Get[string]("test", ctx, "test", time.Second))
cc := NewMapCache[string, string](xx[string, string]{m: map[string]string{}}, nil, func(ctx2 context.Context, k string, a ...any) (string, error) {
NewMapCache[string, string](xx[string, string]{m: map[string]string{}}, nil, func(ctx2 context.Context, k string, a ...any) (string, error) {
fmt.Println("other cache drives. eg: redis,file.....")
return strings.Repeat(k, 2), nil
}, "kkk", time.Hour)
}, "test", time.Hour)
if err := SetMapCache("test", cc); err != nil {
if err := SetMapCache("kkk", x); err != nil {
t.Errorf("SetMapCache() error = %v, wantErr %v", err, nil)
}
fmt.Println(Get[string]("test", ctx, "test", time.Second))
@ -152,16 +152,16 @@ func (x xx[K, V]) ClearExpired(ctx context.Context) {
func TestSetVarCache(t *testing.T) {
t.Run("t1", func(t *testing.T) {
NewVarMemoryCache(func(ctx2 context.Context, a ...any) (string, error) {
bak := NewVarMemoryCache(func(ctx2 context.Context, a ...any) (string, error) {
fmt.Println("memory cache")
return "xxx", nil
}, time.Hour, "test")
fmt.Println(GetVarVal[string]("test", ctx, time.Second))
o := NewVarCache[string](oo[string]{}, func(ctx2 context.Context, a ...any) (string, error) {
NewVarCache[string](oo[string]{}, func(ctx2 context.Context, a ...any) (string, error) {
fmt.Println("other cache drives. eg: redis,file.....")
return "ooo", nil
})
if err := SetVarCache("test", o); err != nil {
}, "test")
if err := SetVarCache("xx", bak); err != nil {
t.Errorf("SetVarCache() error = %v, wantErr %v", err, nil)
}
fmt.Println(GetVarVal[string]("test", ctx, time.Second))

89
cache/map.go vendored
View File

@ -4,6 +4,8 @@ import (
"context"
"errors"
"fmt"
"github.com/fthvgb1/wp-go/cache/reload"
"github.com/fthvgb1/wp-go/helper"
"github.com/fthvgb1/wp-go/helper/maps"
"sync"
"time"
@ -16,24 +18,37 @@ type MapCache[K comparable, V any] struct {
batchCacheFn MapBatchFn[K, V]
getCacheBatch func(c context.Context, key []K, timeout time.Duration, params ...any) ([]V, error)
getCacheBatchToMap func(c context.Context, key []K, timeout time.Duration, params ...any) (map[K]V, error)
increaseUpdate IncreaseUpdate[K, V]
refresh Refresh[K, V]
}
type IncreaseUpdate[K comparable, V any] struct {
CycleTime func() time.Duration
Fn IncreaseFn[K, V]
}
func NewIncreaseUpdate[K comparable, V any](name string, fn IncreaseFn[K, V], cycleTime time.Duration, tFn func() time.Duration) IncreaseUpdate[K, V] {
tFn = reload.FnVal(name, cycleTime, tFn)
return IncreaseUpdate[K, V]{CycleTime: tFn, Fn: fn}
}
type MapSingleFn[K, V any] func(context.Context, K, ...any) (V, error)
type MapBatchFn[K comparable, V any] func(context.Context, []K, ...any) (map[K]V, error)
type IncreaseFn[K comparable, V any] func(c context.Context, currentData V, k K, t time.Time, a ...any) (data V, save bool, refresh bool, err error)
func NewMapCache[K comparable, V any](data Cache[K, V], cacheFunc MapSingleFn[K, V], batchCacheFn MapBatchFn[K, V]) *MapCache[K, V] {
func NewMapCache[K comparable, V any](ca Cache[K, V], cacheFunc MapSingleFn[K, V], batchCacheFn MapBatchFn[K, V], inc IncreaseUpdate[K, V]) *MapCache[K, V] {
r := &MapCache[K, V]{
Cache: data,
Cache: ca,
mux: sync.Mutex{},
cacheFunc: cacheFunc,
batchCacheFn: batchCacheFn,
increaseUpdate: inc,
}
if cacheFunc == nil && batchCacheFn != nil {
r.setDefaultCacheFn(batchCacheFn)
} else if batchCacheFn == nil && cacheFunc != nil {
r.SetDefaultBatchFunc(cacheFunc)
}
ex, ok := any(data).(Expend[K, V])
ex, ok := any(ca).(Expend[K, V])
if !ok {
r.getCacheBatch = r.getCacheBatchs
r.getCacheBatchToMap = r.getBatchToMapes
@ -41,6 +56,10 @@ func NewMapCache[K comparable, V any](data Cache[K, V], cacheFunc MapSingleFn[K,
r.getCacheBatch = r.getBatches(ex)
r.getCacheBatchToMap = r.getBatchToMap(ex)
}
re, ok := any(ca).(Refresh[K, V])
if ok {
r.refresh = re
}
return r
}
@ -101,10 +120,44 @@ func (m *MapCache[K, V]) Flush(ctx context.Context) {
func (m *MapCache[K, V]) GetCache(c context.Context, key K, timeout time.Duration, params ...any) (V, error) {
data, ok := m.Get(c, key)
if ok {
return data, nil
}
var err error
if ok {
if m.increaseUpdate.Fn == nil || m.refresh == nil {
return data, err
}
nowTime := time.Now()
if nowTime.Sub(m.GetLastSetTime(c, key)) < m.increaseUpdate.CycleTime() {
return data, err
}
fn := func() {
m.mux.Lock()
defer m.mux.Unlock()
if nowTime.Sub(m.GetLastSetTime(c, key)) < m.increaseUpdate.CycleTime() {
return
}
dat, save, refresh, er := m.increaseUpdate.Fn(c, data, key, m.GetLastSetTime(c, key), params...)
if er != nil {
err = er
return
}
if refresh {
m.refresh.Refresh(c, key, params...)
}
if save {
m.Set(c, key, dat)
data = dat
}
}
if timeout > 0 {
er := helper.RunFnWithTimeout(c, timeout, fn, fmt.Sprintf("increateUpdate cache %v err", key))
if err == nil && er != nil {
return data, er
}
} else {
fn()
}
return data, err
}
call := func() {
m.mux.Lock()
defer m.mux.Unlock()
@ -118,19 +171,9 @@ func (m *MapCache[K, V]) GetCache(c context.Context, key K, timeout time.Duratio
m.Set(c, key, data)
}
if timeout > 0 {
ctx, cancel := context.WithTimeout(c, timeout)
defer cancel()
done := make(chan struct{}, 1)
go func() {
call()
done <- struct{}{}
}()
select {
case <-ctx.Done():
err = errors.New(fmt.Sprintf("get cache %v %s", key, ctx.Err().Error()))
var vv V
return vv, err
case <-done:
er := helper.RunFnWithTimeout(c, timeout, call, fmt.Sprintf("get cache %v ", key))
if err == nil && er != nil {
err = er
}
} else {
call()
@ -186,7 +229,7 @@ func (m *MapCache[K, V]) getBatchToMap(e Expend[K, V]) func(c context.Context, k
r, er := m.batchCacheFn(ctx, maps.FilterToSlice(needIndex, func(k K, v int) (K, bool) {
return k, true
}), params...)
if err != nil {
if er != nil {
err = er
return
}
@ -253,7 +296,7 @@ func (m *MapCache[K, V]) getBatchToMapes(c context.Context, key []K, timeout tim
}
rr, er := m.batchCacheFn(c, needFlushs, params...)
if err != nil {
if er != nil {
err = er
return
}
@ -318,7 +361,7 @@ func (m *MapCache[K, V]) getCacheBatchs(c context.Context, key []K, timeout time
}
r, er := m.batchCacheFn(c, needFlushs, params...)
if err != nil {
if er != nil {
err = er
return
}
@ -398,7 +441,7 @@ func (m *MapCache[K, V]) getBatches(e Expend[K, V]) func(ctx context.Context, ke
r, er := m.batchCacheFn(ctx, maps.FilterToSlice(needIndex, func(k K, v int) (K, bool) {
return k, true
}), params...)
if err != nil {
if er != nil {
err = er
return
}

View File

@ -2,6 +2,7 @@ package cache
import (
"context"
"github.com/fthvgb1/wp-go/helper"
"github.com/fthvgb1/wp-go/safety"
"time"
)
@ -97,3 +98,13 @@ func (m *MemoryMapCache[K, V]) ClearExpired(_ context.Context) {
return true
})
}
func (m *MemoryMapCache[K, V]) Refresh(_ context.Context, k K, a ...any) {
v, ok := m.Load(k)
if !ok {
return
}
t := helper.ParseArgs(time.Now(), a...)
v.setTime = t
m.Store(k, v)
}

View File

@ -261,7 +261,6 @@ func Push(fn func(), a ...any) {
}
func Reload() {
anyMap.Flush()
callsM.Flush()
flushMapFn.Flush()
callll := calls.Load()
@ -273,3 +272,49 @@ func Reload() {
}
return
}
type Any[T any] struct {
fn func() T
v *safety.Var[T]
isUseManger *safety.Var[bool]
}
func FnVal[T any](name string, t T, fn func() T) func() T {
if fn == nil {
fn = func() T {
return t
}
} else {
t = fn()
}
p := safety.NewVar(t)
e := Any[T]{
fn: fn,
v: p,
isUseManger: safety.NewVar(false),
}
Push(func() {
if !e.isUseManger.Load() {
e.v.Store(fn())
}
})
anyMap.Store(name, e)
return func() T {
return e.v.Load()
}
}
func ChangeFnVal[T any](name string, val T) {
v, ok := anyMap.Load(name)
if !ok {
return
}
vv, ok := v.(Any[T])
if !ok {
return
}
if !vv.isUseManger.Load() {
vv.isUseManger.Store(true)
}
vv.v.Store(val)
}

63
cache/vars.go vendored
View File

@ -2,8 +2,7 @@ package cache
import (
"context"
"errors"
"fmt"
"github.com/fthvgb1/wp-go/helper"
"github.com/fthvgb1/wp-go/safety"
"sync"
"time"
@ -13,14 +12,52 @@ type VarCache[T any] struct {
AnyCache[T]
setCacheFunc func(context.Context, ...any) (T, error)
mutex sync.Mutex
increaseUpdate IncreaseUpdateVar[T]
refresh RefreshVar[T]
}
type IncreaseUpdateVar[T any] struct {
CycleTime func() time.Duration
Fn IncreaseVarFn[T]
}
type IncreaseVarFn[T any] func(c context.Context, currentData T, t time.Time, a ...any) (data T, save bool, refresh bool, err error)
func (t *VarCache[T]) GetCache(ctx context.Context, timeout time.Duration, params ...any) (T, error) {
data, ok := t.Get(ctx)
var err error
if ok {
if t.increaseUpdate.Fn != nil && t.refresh != nil {
nowTime := time.Now()
if t.increaseUpdate.CycleTime() > nowTime.Sub(t.GetLastSetTime(ctx)) {
return data, nil
}
fn := func() {
t.mutex.Lock()
defer t.mutex.Unlock()
da, save, refresh, er := t.increaseUpdate.Fn(ctx, data, t.GetLastSetTime(ctx), params...)
if er != nil {
err = er
return
}
if save {
t.Set(ctx, da)
}
if refresh {
t.refresh.Refresh(ctx, params...)
}
}
if timeout > 0 {
er := helper.RunFnWithTimeout(ctx, timeout, fn, "increaseUpdate cache fail")
if err == nil && er != nil {
err = er
}
} else {
fn()
}
}
return data, nil
}
var err error
call := func() {
t.mutex.Lock()
defer t.mutex.Unlock()
@ -38,19 +75,9 @@ func (t *VarCache[T]) GetCache(ctx context.Context, timeout time.Duration, param
data = r
}
if timeout > 0 {
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
done := make(chan struct{}, 1)
go func() {
call()
done <- struct{}{}
close(done)
}()
select {
case <-ctx.Done():
err = errors.New(fmt.Sprintf("get cache %s", ctx.Err().Error()))
case <-done:
er := helper.RunFnWithTimeout(ctx, timeout, call, "get cache fail")
if err == nil && er != nil {
err = er
}
} else {
call()
@ -98,9 +125,11 @@ func (c *VarMemoryCache[T]) GetLastSetTime(_ context.Context) time.Time {
return c.v.Load().setTime
}
func NewVarCache[T any](cache AnyCache[T], fn func(context.Context, ...any) (T, error)) *VarCache[T] {
func NewVarCache[T any](cache AnyCache[T], fn func(context.Context, ...any) (T, error), inc IncreaseUpdateVar[T], ref RefreshVar[T]) *VarCache[T] {
return &VarCache[T]{
AnyCache: cache, setCacheFunc: fn, mutex: sync.Mutex{},
increaseUpdate: inc,
refresh: ref,
}
}

View File

@ -58,6 +58,8 @@ cacheTime:
userInfoCacheTime: 24h
# 单独评论缓存时间
commentsCacheTime: 24h
# 评论增量更新时间
commentsIncreaseUpdateTime: 30s
# 随机sleep时间
sleepTime: [ 1s,3s ]
# 摘要字数 >0截取指定字数 =0输出出空字符 <0为不截取,原样输出

View File

@ -2,6 +2,7 @@ package helper
import (
"context"
"errors"
"fmt"
str "github.com/fthvgb1/wp-go/helper/strings"
"net/url"
@ -9,6 +10,7 @@ import (
"reflect"
"strconv"
"strings"
"time"
)
func ToAny[T any](v T) any {
@ -144,3 +146,45 @@ func ParseArgs[T any](defaults T, a ...any) T {
}
return defaults
}
func RunFnWithTimeout(ctx context.Context, t time.Duration, call func(), a ...any) (err error) {
ctx, cancel := context.WithTimeout(ctx, t)
defer cancel()
done := make(chan struct{}, 1)
go func() {
call()
done <- struct{}{}
}()
select {
case <-ctx.Done():
msg := ParseArgs("", a...)
if msg != "" {
return errors.New(str.Join(msg, ":", ctx.Err().Error()))
}
return ctx.Err()
case <-done:
close(done)
}
return nil
}
func RunFnWithTimeouts[A, V any](ctx context.Context, t time.Duration, ar A, call func(A) (V, error), a ...any) (v V, err error) {
ctx, cancel := context.WithTimeout(ctx, t)
defer cancel()
done := make(chan struct{}, 1)
go func() {
v, err = call(ar)
done <- struct{}{}
}()
select {
case <-ctx.Done():
msg := ParseArgs("", a...)
if msg != "" {
return v, errors.New(str.Join(msg, ":", ctx.Err().Error()))
}
return v, ctx.Err()
case <-done:
close(done)
}
return v, err
}

View File

@ -58,6 +58,15 @@ func Filter[T any](arr []T, fn func(T, int) bool) []T {
}
return r
}
func Filters[T any](arr []T, fn func(T) bool) []T {
var r []T
for _, t := range arr {
if fn(t) {
r = append(r, t)
}
}
return r
}
func Reduce[R, T any](arr []T, fn func(T, R) R, r R) R {
for _, t := range arr {
@ -143,13 +152,11 @@ func Slice[T any](arr []T, offset, length int) (r []T) {
length = l - offset
}
if l > offset && l >= offset+length {
r = append(make([]T, 0, length), arr[offset:offset+length]...)
arr = append(arr[:offset], arr[offset+length:]...)
r = arr[offset : offset+length]
} else if l <= offset {
return
} else if l > offset && l < offset+length {
r = append(make([]T, 0, length), arr[offset:]...)
arr = arr[:offset]
r = arr[offset:]
}
return
}

View File

@ -8,6 +8,10 @@ func Splice[T any](a *[]T, offset, length int, replacement []T) []T {
}
if offset >= 0 {
if offset+length > l {
if offset == 0 {
*a = []T{}
return arr[:l]
}
return nil
} else if l > offset && l < offset+length {
v := arr[offset:l]