当前位置：首页 > news >正文

【redis 延时队列】使用go-redis的list做异步，生产消费者模式

news 2025/8/8 5:02:52

分享一个用到的，使用go-redis的list做异步，生产消费者模式，接着再用 go 协程去检测队列里是否有东西去消费

如果队列为空，就会一直pop，空轮询导致 cpu 资源浪费和redis qps无效升高，所以可以通过 time.Second 1 秒，降低cpu能耗，和redis的qps

而 BLPop 或者用 BRPop 则是阻塞读

睡眠会导致延迟增大，因为最大可能延迟 1 s 麻，所以阻塞都意思就是一旦来了就立刻醒过来，延迟几乎为 0

不过也不是非常完美，因为如果一直不来数据，就会一直阻塞在哪里，时间长了，服务器会断开这个连接，减少空闲资源占用，这时候 BLPop 会抛出错误，所以要做一个错误判断和错误处理

func (m *RecordMessageListService) Produce(record cdfield.RecordMessageList) {recordJson, _ := json.Marshal(record)global.GSD_REDIS.RPush(context.Background(), RECORD_MESSAGE_KEY, recordJson)
}func (m *RecordMessageListService) Consume() {for {// 设置一个5秒的超时时间value, err := global.GSD_REDIS.BLPop(context.Background(), 5*time.Second, RECORD_MESSAGE_KEY).Result()if err != nil {// 查询出错time.Sleep(1 * time.Second)continue}var record cdfield.RecordMessageList_ = json.Unmarshal([]byte(value[1]), &record)if record.TableName == "glry" {//消费消息glryRecord := cdfield.CdGlryRecord{Name:           record.Name,DocumentNumber: record.IdCard,CreatedAt:      record.CreateTime,}valid, res, _ := utils.GetHNMsg(record.Name, record.IdCard, "")if valid {glryRecord.HsTime = res.YwSamplingTime // YwSamplingTime 检验时间glryRecord.HsResult = res.YwCheckStatusglryRecord.HasHsjl = 1}err = glryRecordService.UpdateHesuan(glryRecord)if err != nil {global.GSD_LOG.Error("消费失败", zap.Error(err))continue}} else if record.TableName == "gzry" {//工作人员消费消息workerRecord := cdfield.CdWorkerRecord{Gzrxm:     record.Name,Gzrsfz:    record.IdCard,CreatedAt: record.CreateTime,}valid, res, _ := utils.GetHNMsg(record.Name, record.IdCard, "")if valid {workerRecord.Hsjl = 1workerRecord.Hsjg = res.YwCheckStatusworkerRecord.Hssj = res.YwSamplingTime}err = cdWorkerService.UpdateWorkerMessageQueue(workerRecord)if err != nil {global.GSD_LOG.Error("消费失败", zap.Error(err))continue}}global.GSD_LOG.Info("消费成功", zap.Any("消费到数据：", value), zap.Any("当前时间是：", time.Now()))time.Sleep(time.Second)}
}