cachedsql.go 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122
  1. package sqlc
  2. import (
  3. "database/sql"
  4. "time"
  5. "github.com/tal-tech/go-zero/core/stores/cache"
  6. "github.com/tal-tech/go-zero/core/stores/redis"
  7. "github.com/tal-tech/go-zero/core/stores/sqlx"
  8. "github.com/tal-tech/go-zero/core/syncx"
  9. )
  10. // see doc/sql-cache.md
  11. const cacheSafeGapBetweenIndexAndPrimary = time.Second * 5
  12. var (
  13. ErrNotFound = sqlx.ErrNotFound
  14. // can't use one SharedCalls per conn, because multiple conns may share the same cache key.
  15. exclusiveCalls = syncx.NewSharedCalls()
  16. stats = cache.NewCacheStat("sqlc")
  17. )
  18. type (
  19. ExecFn func(conn sqlx.SqlConn) (sql.Result, error)
  20. IndexQueryFn func(conn sqlx.SqlConn, v interface{}) (interface{}, error)
  21. PrimaryQueryFn func(conn sqlx.SqlConn, v, primary interface{}) error
  22. QueryFn func(conn sqlx.SqlConn, v interface{}) error
  23. CachedConn struct {
  24. db sqlx.SqlConn
  25. cache cache.Cache
  26. }
  27. )
  28. func NewNodeConn(db sqlx.SqlConn, rds *redis.Redis, opts ...cache.Option) CachedConn {
  29. return CachedConn{
  30. db: db,
  31. cache: cache.NewNode(rds, exclusiveCalls, stats, sql.ErrNoRows, opts...),
  32. }
  33. }
  34. func NewConn(db sqlx.SqlConn, c cache.CacheConf, opts ...cache.Option) CachedConn {
  35. return CachedConn{
  36. db: db,
  37. cache: cache.New(c, exclusiveCalls, stats, sql.ErrNoRows, opts...),
  38. }
  39. }
  40. func (cc CachedConn) DelCache(keys ...string) error {
  41. return cc.cache.Del(keys...)
  42. }
  43. func (cc CachedConn) GetCache(key string, v interface{}) error {
  44. return cc.cache.Get(key, v)
  45. }
  46. func (cc CachedConn) Exec(exec ExecFn, keys ...string) (sql.Result, error) {
  47. res, err := exec(cc.db)
  48. if err != nil {
  49. return nil, err
  50. }
  51. if err := cc.DelCache(keys...); err != nil {
  52. return nil, err
  53. }
  54. return res, nil
  55. }
  56. func (cc CachedConn) ExecNoCache(q string, args ...interface{}) (sql.Result, error) {
  57. return cc.db.Exec(q, args...)
  58. }
  59. func (cc CachedConn) QueryRow(v interface{}, key string, query QueryFn) error {
  60. return cc.cache.Take(v, key, func(v interface{}) error {
  61. return query(cc.db, v)
  62. })
  63. }
  64. func (cc CachedConn) QueryRowIndex(v interface{}, key string, keyer func(primary interface{}) string,
  65. indexQuery IndexQueryFn, primaryQuery PrimaryQueryFn) error {
  66. var primaryKey interface{}
  67. var found bool
  68. if err := cc.cache.TakeWithExpire(&primaryKey, key, func(val interface{}, expire time.Duration) (err error) {
  69. primaryKey, err = indexQuery(cc.db, v)
  70. if err != nil {
  71. return
  72. }
  73. found = true
  74. return cc.cache.SetWithExpire(keyer(primaryKey), v, expire+cacheSafeGapBetweenIndexAndPrimary)
  75. }); err != nil {
  76. return err
  77. }
  78. if found {
  79. return nil
  80. }
  81. return cc.cache.Take(v, keyer(primaryKey), func(v interface{}) error {
  82. return primaryQuery(cc.db, v, primaryKey)
  83. })
  84. }
  85. func (cc CachedConn) QueryRowNoCache(v interface{}, q string, args ...interface{}) error {
  86. return cc.db.QueryRow(v, q, args...)
  87. }
  88. // QueryRowsNoCache doesn't use cache, because it might cause consistency problem.
  89. func (cc CachedConn) QueryRowsNoCache(v interface{}, q string, args ...interface{}) error {
  90. return cc.db.QueryRows(v, q, args...)
  91. }
  92. func (cc CachedConn) SetCache(key string, v interface{}) error {
  93. return cc.cache.Set(key, v)
  94. }
  95. func (cc CachedConn) Transact(fn func(sqlx.Session) error) error {
  96. return cc.db.Transact(fn)
  97. }