goredis.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477
  1. package redis
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "hash/crc32"
  7. "log"
  8. "math"
  9. "regexp"
  10. "strconv"
  11. "strings"
  12. "time"
  13. "github.com/go-redis/redis/v8"
  14. )
  15. type GoRedis struct {
  16. //Password string
  17. Code string
  18. DB int
  19. HashDb bool //是否是多个数据库
  20. DBS []int //数据库列表
  21. Ctx context.Context
  22. CMap map[int]*redis.Client
  23. BakDb bool //是否有备用节点
  24. Bak *GoRedis //备用节点连接
  25. }
  26. //初始化单个"[other=]127.0.0.1:2203[|#]127.0.0.1:2204[/0]=0-1=1-10=300" [代码]、地址[|备用地址[/代表默认库][#多节点地址,默认库都为0]、库、最大池、空闲时间默认300秒
  27. /*
  28. 示例
  29. 127.0.0.1:2203
  30. 127.0.0.1:2203=2=15 默认库为2,最大连接为15,最小连接是0.2*15为3
  31. other=127.0.0.1:2203 带code配置
  32. other=127.0.0.1:2203=0=10
  33. other=127.0.0.1:2203=0=10=300 空闲时间300秒
  34. other=127.0.0.1:2203=0-8=10=300 带code配置,使用多个库即0 1 2 3 4 5 6 7 8,最大连接为10,最小连接是0.2*10为2,空闲时间300秒
  35. other=127.0.0.1:2203=0-8=1-10=300 带code配置,使用多个库即0 1 2 3 4 5 6 7 8,最大连接为10,最小连接1,空闲时间300秒
  36. other=127.0.0.1:2203|127.0.0.1:2204=0-8=10=300 带code配置,使用备用节点(仅供取数据时使用),使用多个库即0 1 2 3 4 5 6 7 8,最大连接为10,最小连接1,空闲时间300秒
  37. other=127.0.0.1:2203|127.0.0.1:2204/0=0-8=10=300 带code配置,使用备用节点(仅供取数据时使用,备用节点默认库为0),使用多个库即0 1 2 3 4 5 6 7 8,最大连接为10,最小连接1,空闲时间300秒
  38. other=127.0.0.1:2203#127.0.0.1:2204#127.0.0.1:2205#127.0.0.1:2206=0=10=300 带code配置,使用多节点模式(多节点所有默认库为0),最大连接为10,最小连接2,空闲时间300秒
  39. 注意:
  40. 1、当使用多节点时,用正则获取keys时,再用Get取模获取数据时,一定要保障放入和取出是同一个hashcode,否则数据永远取不到
  41. */
  42. //解析配置并初始化
  43. func (r *GoRedis) Init(opt interface{}) {
  44. check := false
  45. code, addr, dbs, pool, idle := "", "", []int{0}, []int{2, 30}, 300
  46. if so, ok := opt.(string); ok {
  47. arr := strings.Split(so, "=")
  48. regAddr := regexp.MustCompile("[0-9.a-zA-Z/]+:[0-9]+.*")
  49. if len(arr) == 1 {
  50. if regAddr.MatchString(arr[0]) {
  51. check = true
  52. addr = arr[0]
  53. }
  54. } else if len(arr) > 1 {
  55. index := 0
  56. if regAddr.MatchString(arr[0]) {
  57. index = 1
  58. addr = arr[0]
  59. check = true
  60. } else if regAddr.MatchString(arr[1]) {
  61. check = true
  62. addr = arr[1]
  63. code = arr[0]
  64. }
  65. //解析库配置
  66. if len(arr) > 2-index { //dbs配置
  67. dbs1 := strings.Split(arr[2-index], "-")
  68. if len(dbs1) == 1 || len(dbs1) == 2 {
  69. check = true
  70. dbs[0], _ = strconv.Atoi(dbs1[0])
  71. if len(dbs1) == 2 {
  72. tmp, _ := strconv.Atoi(dbs1[1])
  73. dbs = append(dbs, tmp)
  74. }
  75. } else {
  76. check = false
  77. }
  78. //解析连接池配置
  79. if len(arr) > 3-index {
  80. pool1 := strings.Split(arr[3-index], "-")
  81. if len(pool1) == 1 || len(pool1) == 2 {
  82. check = true
  83. if len(pool1) == 1 {
  84. pool[1], _ = strconv.Atoi(pool1[0])
  85. pool[0] = int(math.Ceil(float64(pool[1]) * 0.2))
  86. } else {
  87. pool[0], _ = strconv.Atoi(pool1[0])
  88. pool[1], _ = strconv.Atoi(pool1[1])
  89. }
  90. } else {
  91. check = false
  92. }
  93. //解析最大空闲时间配置
  94. if len(arr) > 4-index {
  95. idle, _ = strconv.Atoi(arr[4-index])
  96. if idle == 0 {
  97. idle = 300
  98. }
  99. }
  100. }
  101. }
  102. }
  103. } else if _, ok := opt.(map[string]interface{}); ok {
  104. }
  105. if check {
  106. addrs := strings.Split(addr, "|")
  107. if len(addrs) == 2 { //备用节点的模式 2选1
  108. r.init(addrs[0], code, dbs, pool[1], pool[0], idle)
  109. r.BakDb = true
  110. //有备用节点,集群先不考虑,支持指定的库
  111. addr1 := strings.Split(addrs[1], "/")
  112. r.Bak = &GoRedis{}
  113. if len(addr1) == 2 { //没有指定库
  114. i, _ := strconv.Atoi(addr1[1])
  115. dbs = []int{i}
  116. }
  117. r.Bak.init(addr1[0], code, dbs, pool[1], pool[0], idle)
  118. } else if len(addrs) == 1 {
  119. addr1 := strings.Split(addrs[0], "#") //是多节点模式,所有库默认为0
  120. if len(addr1) == 1 {
  121. r.init(addr1[0], code, dbs, pool[1], pool[0], idle)
  122. } else {
  123. r.Code = code
  124. r.DB = 0
  125. r.Ctx = context.Background()
  126. r.CMap = map[int]*redis.Client{}
  127. r.HashDb = true
  128. r.DBS = []int{}
  129. for i := 0; i < len(addr1); i++ {
  130. r.DBS = append(r.DBS, i)
  131. r.CMap[i] = redis.NewClient(&redis.Options{
  132. Addr: addr1[i],
  133. PoolSize: pool[1],
  134. MinIdleConns: pool[0],
  135. IdleTimeout: time.Duration(idle) * time.Second,
  136. })
  137. }
  138. }
  139. }
  140. }
  141. log.Println(check, code, addr, dbs, pool, idle, r.DBS)
  142. }
  143. //初始化连接
  144. func (r *GoRedis) init(addr, code string, dbs []int, poolSize, minIdleConns, idleTimeOut int) {
  145. r.Code = code
  146. r.DB = dbs[0]
  147. r.Ctx = context.Background()
  148. r.CMap = map[int]*redis.Client{}
  149. if len(dbs) > 1 {
  150. r.HashDb = true
  151. r.DBS = []int{}
  152. for i := dbs[0]; i <= dbs[1]; i++ {
  153. r.DBS = append(r.DBS, i)
  154. }
  155. } else {
  156. r.DBS = dbs
  157. }
  158. for k, v := range r.DBS {
  159. r.CMap[k] = redis.NewClient(&redis.Options{
  160. Addr: addr,
  161. DB: v,
  162. PoolSize: poolSize,
  163. MinIdleConns: minIdleConns,
  164. IdleTimeout: time.Duration(idleTimeOut) * time.Second,
  165. })
  166. }
  167. }
  168. func D(t int) time.Duration {
  169. return time.Duration(t) * time.Second
  170. }
  171. //取db
  172. func (r *GoRedis) GetDB(key string) int {
  173. if r.HashDb {
  174. return hashCode(key) % len(r.DBS)
  175. } else {
  176. return r.DB
  177. }
  178. }
  179. //根据key取hash
  180. func hashCode(key string) int {
  181. v := int(crc32.ChecksumIEEE([]byte(key)))
  182. if v < 0 {
  183. v = -v
  184. }
  185. return v
  186. }
  187. //-----具体方法
  188. //简单的Put方法
  189. func (r *GoRedis) Put(key string, val interface{}) (string, error) {
  190. stutsCmd := r.CMap[r.GetDB(key)].Set(r.Ctx, key, val, 0)
  191. str, err := stutsCmd.Result()
  192. return str, err
  193. }
  194. func (r *GoRedis) Set(key string, val interface{}, timeout int) (string, error) {
  195. stutsCmd := r.CMap[r.GetDB(key)].Set(r.Ctx, key, val, D(timeout))
  196. //cmd := r.CMap[r.GetDB(key)].Do(r.Ctx, "setex", key, timeout, val)
  197. str, err := stutsCmd.Result()
  198. return str, err
  199. }
  200. //简单的Get方法,无结果返回空串,err: redis: nil
  201. func (r *GoRedis) Get(key string) (string, error) {
  202. stutsCmd := r.CMap[r.GetDB(key)].Get(r.Ctx, key)
  203. str, err := stutsCmd.Result()
  204. // log.Println("1", str, "-", err)
  205. if err != nil {
  206. //有备用节点
  207. if r.BakDb {
  208. str, err = r.Bak.Get(key)
  209. }
  210. }
  211. return str, err
  212. }
  213. //根据正则取key,未考虑负载、多库
  214. func (r *GoRedis) GetByPattern(key string) (res []string, err error) {
  215. serr := ""
  216. for _, v := range r.CMap {
  217. strSlice := v.Keys(r.Ctx, key)
  218. arr, err1 := strSlice.Result()
  219. if len(arr) > 0 {
  220. res = append(res, arr...)
  221. } else if err1 != nil {
  222. serr += err1.Error()
  223. }
  224. }
  225. if serr != "" {
  226. err = errors.New(serr)
  227. }
  228. return
  229. }
  230. //批量保存数据
  231. // 不支持- MSet("key1", "value1", "key2", "value2")
  232. // 不支持- MSet([]string{"key1", "value1", "key2", "value2"})
  233. // []interface{[]interface{k1,v1},[]interface{k2,v2},[]interface{k3,v3} }
  234. // - MSet(map[string]interface{}{"key1": "value1", "key2": "value2"})
  235. func (r *GoRedis) BulkPut(timeout int, obj interface{}) {
  236. if r.HashDb {
  237. if timeout < 1 {
  238. timeout = 0
  239. }
  240. timeEx := time.Duration(timeout) * time.Second
  241. switch obj.(type) {
  242. case []interface{}:
  243. if objs, ok := obj.([]interface{}); ok {
  244. for _, _tmp := range objs {
  245. tmp, ok := _tmp.([]interface{})
  246. if ok && len(tmp) == 2 {
  247. key, ok1 := tmp[0].(string)
  248. if ok1 {
  249. r.CMap[r.GetDB(key)].Set(r.Ctx, key, tmp[1], timeEx)
  250. }
  251. }
  252. }
  253. }
  254. case map[string]interface{}:
  255. if objs, ok := obj.(map[string]interface{}); ok {
  256. for k, v := range objs {
  257. r.CMap[r.GetDB(k)].Set(r.Ctx, k, v, timeEx)
  258. }
  259. }
  260. }
  261. } else {
  262. //单库直接保存
  263. if timeout < 1 {
  264. stutsCmd := r.CMap[r.DB].MSet(r.Ctx, obj)
  265. str, err := stutsCmd.Result()
  266. log.Println(str, err)
  267. } else {
  268. timeEx := time.Duration(timeout) * time.Second
  269. //设置超时
  270. lenth := 0
  271. cmds, err := r.CMap[r.DB].Pipelined(r.Ctx, func(pipe redis.Pipeliner) error {
  272. if objs, ok := obj.([]interface{}); ok {
  273. lenth = len(objs)
  274. for _, _tmp := range objs {
  275. tmp, ok := _tmp.([]interface{})
  276. if ok && len(tmp) == 2 {
  277. key, ok1 := tmp[0].(string)
  278. if ok1 {
  279. pipe.Set(r.Ctx, key, tmp[1], timeEx)
  280. }
  281. }
  282. }
  283. } else if objs, ok := obj.(map[string]interface{}); ok {
  284. lenth = len(objs)
  285. for k, v := range objs {
  286. pipe.Set(r.Ctx, k, v, timeEx)
  287. }
  288. } else {
  289. log.Println("bulkPut type error")
  290. }
  291. return nil
  292. })
  293. if err != nil {
  294. log.Println("bulkPut error", err.Error())
  295. }
  296. if len(cmds) == lenth { //数据相等
  297. for _, v := range cmds {
  298. log.Println("cmd", v.Args(), v.String())
  299. }
  300. }
  301. }
  302. }
  303. }
  304. //设置超时时间,单位秒
  305. func (r *GoRedis) SetExpire(key string, expire int) error {
  306. boolCmd := r.CMap[r.GetDB(key)].Expire(r.Ctx, key, D(expire))
  307. return boolCmd.Err()
  308. }
  309. //判断一个key是否存在
  310. func (r *GoRedis) Exists(key string) bool {
  311. intCmd := r.CMap[r.GetDB(key)].Exists(r.Ctx, key)
  312. i, err := intCmd.Result()
  313. if err != nil {
  314. log.Println("redisutil-exists", key, err)
  315. }
  316. return i == 1
  317. }
  318. //直接返回字节流
  319. func (r *GoRedis) GetBytes(key string) (ret *[]byte, err error) {
  320. cmd := r.CMap[r.GetDB(key)].Do(r.Ctx, "GET", key)
  321. res, err := cmd.Result()
  322. log.Println(res)
  323. if err != nil {
  324. log.Println("redisutil-GetBytesError", err)
  325. } else {
  326. if tmp, ok := res.([]byte); ok {
  327. ret = &tmp
  328. } else if tmp, ok := res.(string); ok {
  329. bs := []byte(tmp)
  330. ret = &bs
  331. } else {
  332. err = errors.New("redis返回数据格式不对")
  333. }
  334. }
  335. return
  336. }
  337. //支持删除多个key
  338. func (r *GoRedis) Del(key ...string) (b bool) {
  339. i := 0
  340. if r.HashDb {
  341. for _, k := range key {
  342. cmd := r.CMap[r.GetDB(k)].Del(r.Ctx, k)
  343. i1, _ := cmd.Result()
  344. i += int(i1)
  345. }
  346. } else {
  347. intCmd := r.CMap[r.DB].Del(r.Ctx, key...)
  348. i1, _ := intCmd.Result()
  349. i = int(i1)
  350. }
  351. return i == len(key)
  352. }
  353. //根据代码和前辍key删除多个
  354. func (r *GoRedis) DelByPattern(key string) {
  355. res, _ := r.GetByPattern(key)
  356. if len(res) > 0 {
  357. r.Del(res...)
  358. }
  359. }
  360. //自增计数器
  361. func (r *GoRedis) Incr(key string) (int64, error) {
  362. intCmd := r.CMap[r.GetDB(key)].Incr(r.Ctx, key)
  363. i, err := intCmd.Result()
  364. return i, err
  365. }
  366. //自减
  367. func (r *GoRedis) Decrby(key string, val int) (int64, error) {
  368. intCmd := r.CMap[r.GetDB(key)].DecrBy(r.Ctx, key, int64(val))
  369. i, err := intCmd.Result()
  370. return i, err
  371. }
  372. //批量取多个key
  373. func (r *GoRedis) Mget(key []string) []interface{} {
  374. if r.HashDb {
  375. mdb := map[int][]string{}
  376. for _, v := range key { //分组
  377. arr := mdb[r.GetDB(v)]
  378. if arr == nil {
  379. arr = []string{v}
  380. } else {
  381. arr = append(arr, v)
  382. }
  383. mdb[r.GetDB(v)] = arr
  384. }
  385. res := []interface{}{}
  386. for k, v := range mdb {
  387. sliceCmd := r.CMap[k].MGet(r.Ctx, v...)
  388. res1, err := sliceCmd.Result()
  389. if err != nil {
  390. log.Println("Mget error", err)
  391. }
  392. res = append(res, res1...)
  393. }
  394. return res
  395. } else {
  396. sliceCmd := r.CMap[r.DB].MGet(r.Ctx, key...)
  397. res, err := sliceCmd.Result()
  398. if err != nil {
  399. log.Println("Mget error", err)
  400. }
  401. return res
  402. }
  403. }
  404. //取出并删除Key
  405. func (r *GoRedis) Pop(key string) (result interface{}) {
  406. strCmd := r.CMap[r.GetDB(key)].GetDel(r.Ctx, key)
  407. b, err := strCmd.Bytes()
  408. if err != nil {
  409. log.Println("Poperr bytes", err)
  410. return
  411. }
  412. err1 := json.Unmarshal(b, &result)
  413. if err1 != nil {
  414. log.Println("Poperr json ", err)
  415. return
  416. }
  417. return
  418. }
  419. //list操作
  420. func (r *GoRedis) LPOP(list string) (result interface{}) {
  421. strCmd := r.CMap[r.GetDB(list)].LPop(r.Ctx, list)
  422. b, err := strCmd.Bytes()
  423. if err != nil {
  424. log.Println("LPOP bytes", err)
  425. return
  426. }
  427. err1 := json.Unmarshal(b, &result)
  428. if err1 != nil {
  429. log.Println("LPOP json ", err)
  430. return
  431. }
  432. return
  433. }
  434. func (r *GoRedis) RPUSH(list string, val interface{}) bool {
  435. intCmd := r.CMap[r.GetDB(list)].RPush(r.Ctx, list, val)
  436. i, err := intCmd.Result()
  437. if err != nil {
  438. log.Println("RPUSH bytes", err)
  439. return false
  440. }
  441. return i == 1
  442. }
  443. func (r *GoRedis) LLEN(list string) int64 {
  444. intCmd := r.CMap[r.GetDB(list)].LLen(r.Ctx, list)
  445. i, err := intCmd.Result()
  446. if err != nil {
  447. log.Println("RPUSH bytes", err)
  448. }
  449. return i
  450. }