redis缓存池使用

本是古典何须时尚 2022-12-21 10:50 79阅读 0赞

##### Redis #####

Redis里存储的数据是key、value键值对的格式，如果我们同时有多个请求访问redis拿数据，但同时要保证数据不重复，这里我们采用set形式（value不重复，string类型的key值对应数组类型的value，数组里可放置多个string类型的数据）

package ipv4pool
    
    import (
    	"context"
    	"fmt"
    	"time"
    	"amber/db"
    
    	"github.com/go-redis/redis"
    	"github.com/sirupsen/logrus"
    )
    
    var Pool *ipv4Pool
    
    const (
    	ipv4PoolKey = "ipv4_pool"    //设置redis的key
    	ipv4Capacity   = 255         //设置redis里value的最大长度
    	getTimeout     = 5 * time.Second    //超时时间
    	expire = 24                       //过期时间 
    )    
    
    type ipv4Pool struct { 
    	cap      int
    	client   *redis.Client
    	isUpdate chan bool
    	ctx      context.Context
    }
    
    // 初始化redis数据库
    func NewCache(addr string) (*redis.Client, error) { 
    	c := redis.NewClient(&redis.Options{ 
    		Addr: addr})
    	_, err := c.Ping().Result()
    	if err != nil { 
    		return nil, err
    	}
    	return c, err
    }
    
    // 初始化连接池pool
    func Init(addr string, ctx context.Context) (err error) { 
    	c, err := NewCache(addr)
    	if err != nil { 
    		logrus.WithError(err).Error("can not init ipv4 pool redis client")
    		return err
    	}
    
    	Pool = &ipv4Pool{ 
    		client:   c,
    		cap:      ipv4Capacity,
    		isUpdate: make(chan bool),
    		ctx:      ctx}
    	go Pool.start()  //接收程序异常终止信号，删除整个pool
    	return nil
    }
    
    // 接收程序异常终止信号
    func (ip *ipv4Pool) start() { 
    	defer func() { 
    		if r := recover(); r != nil { 
    			redisKey := ipv4PoolKey
    			p.client.Del(redisKey)
    			logrus.Info("force delete ipv4 pool ", redisKey, " by ", r)
    		}
    	}()
    
    	for { 
    		select { 
    		case <-ip.ctx.Done():
    			ip.DeleteIpv4Pool()
    			return
    
    		}
    	}
    }
    
    // 从redis里获取key对应的value
    func (ip *ipv4Pool) GetIpv4() (string, error) { 
    	var (
    		key      string
    		ipv4     string
    		err      error
    		isUpdate = make(chan bool, 1)
    	)
    	key = ipv4PoolKey
    	
    	// spop随机从redis池里拿出一个ip
    	if ipv4, err = ip.client.SPop(key).Result(); err != nil { 
    		select { 
    		//再次获取，第一次获取时池子里是空的，报错，将数据库的数据add到池子
    		case isUpdate <- true:
    			err = ip.Updater()
    			<-isUpdate
    			if err != nil { 
    				return "", err
    			}
    		case <-time.After(getTimeout):
    			logrus.Warn("update ipv4 pool time out")
    		}
    
    		// 此时从redis拿数据仍报错，就是池子里没有ip
    		if ipv4, err = ip.client.SPop(key).Result(); err != nil { 
    			errInfo := fmt.Sprintf("get ipv4 from redis pool with error %v, may be no more ip left", err)
    			logrus.Error(errInfo)
    			return "", fmt.Errorf(errInfo)
    		}
    	}
    
    	return ipv4, nil
    }
    
    // 单独删除一个ip
    func (ip *ipv4Pool) DeleteIpv4(ipv4 string) error { 
    	redisKey := ipv4PoolKey
    	if _, err := ip.client.SRem(redisKey, ipv4).Result(); err != nil { 
    		logrus.Errorf("remove ipv4: %s from redis pool failure", ipv4)
    		return err
    	}
    	return nil
    }
    // 删除ipv4池
    func (ip *ipv4Pool) DeleteIpv4Pool() { 
    	redisKey := ipv4PoolKey
    	ip.client.Del(redisKey)
    	logrus.Info("force to delete ipv4 pool ", redisKey)
    }
    
    // 查看当前池子的长度
    func (ip *ipv4Pool) len() int { 
    	redisKey := ipv4PoolKey
    	l, err := ip.client.SCard(redisKey).Result()
    	if err != nil { 
    		logrus.Error("can not get length of ipv4 pool")
    		return 0
    	}
    	return int(l)
    }
    
    // 从db获取数据放入redis，刷新连接池
    func (ip *ipv4Pool) Updater() error { 
    	var (
    		err     error
    		errInfo string
    	)
    	redisKey := ipv4PoolKey
    	ipv4s := &[]string{ }
    
    	updateTime := time.Now()
    	defer func() { 
    		logrus.Debugf("update ipv4 pool elapsed: %v...", time.Since(updateTime))
    	}()
    
    	if ip.len() >= ip.cap { 
    		logrus.Info("ipv4 pool is already full")
    		return nil
    	}
    	if err = db.DB.Debug().Table("ipv4_pool").Where("priority = (select max(priority) from ipv4_pool)").Limit(50).Pluck("ip", ipv4s).Error; err != nil { 
    		errInfo = fmt.Sprintf("can not get max priority ipv4 from db with error %v", err)
    		logrus.Error(errInfo)
    		return fmt.Errorf(errInfo)
    	}
    	
    	if len(*ipv4s) == 0 { 
    		errInfo = fmt.Sprintf("no more ipv4s in db")
    		logrus.Warn(errInfo)
    		return fmt.Errorf(errInfo)
    	}
    
    	logrus.Infof("ipv4s %+v", *ipv4s)
    	if _, err = p.client.SAdd(redisKey, *ipv4s).Result(); err != nil { 
    		errInfo = fmt.Sprintf("cannot add ipv4s to redis pool with error %v", err)
    		logrus.Error(errInfo)
    		return fmt.Errorf(errInfo)
    	}
    
    	if p.len() >= p.cap { 
    		logrus.Info("ipv6 pool is full")
    	}
    
    	// 设置过期时间后，redis里key超过设置时间自动删除
    	p.client.Expire(redisKey, time.Duration(expire)*time.Hour)
    	return nil
    }