goredis.go 14 KB

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