我是如何用 redis 分布式锁来解决线上历史业务问题的

这篇具有很好参考价值的文章主要介绍了我是如何用 redis 分布式锁来解决线上历史业务问题的。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

近期发现,开发功能的时候发现了一个 mq 消费顺序错乱(历史遗留问题),导致业务异常的问题,看看我是如何解决的

问题抛出

首先,简单介绍一下情况:

线上 k8s 有多个 pod 会去消费 mq 中的消息,可是生产者发送的消息是期望一定要有序去消费,此时要表达的是,例如 生产者如果发送了 3 个通知消息,分别是

  • 1 系统已经在 / 组下面添加 a 组,你记得绑定策略 (例如 / 组绑定的是策略是:允许看视频类型的网站)
  • 2 系统已经在 /a 组下添加了 b 组, 你记得绑定策略(期望绑定的策略和他的父组策略一样)
  • 3 系统已经在 b 组下面添加 小 d 用户,你的绑定策略(期望绑定的策略和他的所在组一样)

此处,若有 3 个 pod 的分别拿到了上述 3 条消息,但是自身实际消费完毕的顺序可能是 先完成了 3 消息对应的业务逻辑,再是 2 消息 的业务逻辑,最后是 1 消息的业务逻辑

那么这个时候,小 d 用户就没有绑定上 允许看视频类型的网站 这一条策略,自然 b组 和 a 组也没有绑定上这条策略,这就和我们预期的完全不一致了

当然,实际情况对于单条单条的消息处理基本不会出现这种偏差,但是在批量处理的时候,就会出现实际业务处理顺序与期望不一致的情况,那么就是妥妥的线上问题了(小 d 上网的时候想看视频,可是一直看不了,于是就疯狂投诉。。。)

思考解决

对于这个问题如何解决呢?

我们知道,咱们使用 mq 的目的是为了做到去处理我们的异步逻辑,还能对流量进行削峰,服务间解耦

对于咱们的 A 服务,已经处理了关于添加用户的,添加组的逻辑,发送通知消息给到 B 服务的时候,B 服务自身的处理顺序,未按照既定的顺序真实按照顺序消费完毕,导致出现了业务问题

想法一

我们是期望 B 服务团队去添加批量接口,A 服务将需要通知的信息,排序好给到 B 服务,一个整包, B 服务的单个 pod 接收到这个大包,然后按照顺序处理消息即可,但是这个方式弊端比较明显

  • 当发送了多个批量大包消息的时候,B 服务如果自身处理不过来,也会导致类似的问题,无法根治
  • 需要 B 服务新增和修改的代码较多,肯定谈不下来
  • 而且对于绑定策略的服务来说,不仅仅是 B 服务,还有 C 服务,D 服务呢,他们都要改造… 这个想法就。。。

想法二

对于这一个业务,也不能去对整个架构大改,对于这些历史遗留问题,能少动就少动,兄弟们你们都懂的

于是便想出了使用 redis 分布式锁来处理,对于一个部署在 k8s 中服务的多个 pod 去抢占,谁先抢到锁,那么就谁消费 mq 中的消息,没有抢到锁的 pod ,那就过一会再抢

当然,对于其他类型的业务是没有影响的

如何去实现这个想法呢,我们可以模拟一下

我是如何用 redis 分布式锁来解决线上历史业务问题的,redis,分布式,数据库

  • 1 首先,我们设置一个 redis 的 key,例如 [服务名]_lockmq, 值的话咱们就任意设置,默认就用 服务名 做 value 吧,过期时间暂定 30 秒,有需求的可以调大
  • 2 如果设置成功,则处理成功之后的事情

    • 2.1 初始化 mq 消费者,并开启协程进行消费

      • 2.1.1 如果初始化失败,则直接返回,退到第 1 步
    • 2.2 对 redis 锁进行续期,此处咱们 10 秒续期一次

      • 如果续期失败,则直接返回,退到第 1 步
  • 3 若拿锁失败,则休息 10 秒再去拿锁

这样来处理的话,我们就可以应对多个 pod 来消费同一类消息的时候,保证同时只有一个 pod 在处理 mq 中的消息了,当然如果正在处理消息的 pod 出现了异常,对于其他 pod ,最晚会在 40 秒之后拿到锁,对于大量的消息来说,这个还是可以容忍的

对应的代码逻辑如下:

  • 简单连接 redis, redis 分布式锁的主逻辑如下

    • 连接 redis ,DB 默认为 0 号
var rdb = redis.NewClient(&redis.Options{
   Addr:     "localhost:6379",
   Password: "123456",
   DB:       0,
})

func LockMq(svrName string) {
   key := fmt.Sprintf("%s_lockmq", svrName)
   // 尝试加锁
   var set bool
   for {
      set = redisLock(key, svrName, time.Second*30)
      if set {
         log.Println("redisLock success ")
         if err := afterLockSuccess(key); err != nil {
            // 如果此处有err ,自然是 mq 初始化失败
            log.Println("mq init error: ", err)
         }else{
            log.Println("redisLock expire failed ")
         }
         time.Sleep(time.Second * 10)
         continue
      }
      // afterLockFailed()
      log.Println("redisLock failed ")
      time.Sleep(time.Second * 10)
   }
}
  • 基本的加锁实现

    • 设置 key , value , 过期时间为 30 秒
func redisLock(key, value string, duration time.Duration) bool {
   set, err := rdb.SetNX(context.TODO(), key, value, duration).Result()
   if err != nil {
      log.Println("setnx failed, error: ", err)
      return false
   }
   return set
}
  • 加锁成功之后,初始化 mq 客户端并进行消费,续期 redis 分布式锁
func afterLockSuccess(key string) error {
   // 初始化需要做的内容或者句柄
   // xxx
   // 对于此处的初始化 mq 句柄失败才返回 err
   ch := make(chan struct{}, 1)
   go func() {
      // 模拟消费消息
      for {
         select {
         case <-ch:
            log.Println("expire failed,mq close")
            return
         default:
            log.Println("is consuming msg")
            time.Sleep(time.Second * 2)
         }
      }
   }()

   for {
      time.Sleep(time.Second*10)
      // 续期
      set, err := rdb.PExpire(context.TODO(), key, time.Second*30).Result()
      if err != nil {
         log.Println("PExpire error!! ", err)
         return nil
      }
      if !set {
         ch <- struct{}{}
         log.Println("PExpire failed!!")
         return nil
      }
      log.Println("PExpire success!! ")
   }

}

具体的测试直接调用 LockMq 函数即可

func main(){
   go redislock.LockMq("helloworld")
   select{}
}

模拟启动多个 pod 去抢锁,抢到锁的执行业务,继续续期,抢不到锁的休息一会再接着抢

程序 a 先启动,程序 b 后启动

程序 a 日志如下:

程序 a 起来之后,启动一段时间之后,kill 掉 程序 a

我是如何用 redis 分布式锁来解决线上历史业务问题的,redis,分布式,数据库

程序 b 日志如下:

程序 b 先是获取锁失败,过 30s 左右,程序 b 能正常获取到锁

我是如何用 redis 分布式锁来解决线上历史业务问题的,redis,分布式,数据库

关于源码可以查看地址:https://github.com/qingconglaixueit/my_redis_demo

感谢阅读,欢迎交流,点个赞,关注一波 再走吧

欢迎点赞,关注,收藏

朋友们,你的支持和鼓励,是我坚持分享,提高质量的动力

好了,本次就到这里

技术是开放的,我们的心态,更应是开放的。拥抱变化,向阳而生,努力向前行。

我是阿兵云原生,欢迎点赞关注收藏,下次见~

可以进入地址进行体验和学习:https://xxetb.xet.tech/s/3lucCI文章来源地址https://www.toymoban.com/news/detail-707616.html

到了这里,关于我是如何用 redis 分布式锁来解决线上历史业务问题的的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Redis实现分布式锁之----超时和失效(非原子性)问题----解决方案

    Redis实现分布式锁之----超时和失效(非原子性)问题----解决方案 超时和失效(非原子性)问题 原子性问题 :上锁时存入线程名称,删除时要先判断锁内的名称是不是自己的,是再删除,但是后面的判断 和删除非原子性 ,会有并发安全问题。 不可重入问题 :一个线程只能

    2024年02月07日
    浏览(43)
  • Redis实现方式开启新篇章,解决分布式环境下的资源竞争问题,提升系统稳定性

    分布式锁一般有三种实现方式: 数据库乐观锁; 基于Redis的分布式锁; 基于ZooKeeper的分布式锁 本篇博客将介绍第二种方式,基于Redis实现分布式锁。 虽然网上已经有各种介绍Redis分布式锁实现的博客,然而他们的实现却有着各种各样的问题,为了避免误人子弟,本篇博客将

    2024年02月07日
    浏览(45)
  • Redis与分布式-分布式锁

    接上文 Redis与分布式-集群搭建 为了解决上述问题,可以利用分布式锁来实现。 重新复制一份redis,配置文件都是刚下载时候的不用更改,然后启动redis服务和redis客户。 redis存在这样的命令:和set命令差不多,但是它有一个机制,当指定的key不存在的时候,才能进行插入,实

    2024年02月07日
    浏览(50)
  • Redis分布式锁和分布式事务

    Redis分布式锁和分布式事务 一、Redis分布式锁 1.1 watch和事务实现分布式锁 原理是通过watch来观察一个变量,一个线程在操作的时候,其他线程会操作失败,相当于乐观锁。 1.2 setnx实现分布式锁 原理是通过setnx设置一个变量,设置成功的线程抢到锁,执行相关的业务,执行完毕

    2024年02月09日
    浏览(45)
  • 【redis】redis分布式锁

    一、为什么需要分布式锁 1.在java单机服务中,jvm内部有一个全局的锁监视器,只有一个线程能获取到锁,可以实现线程之间的互斥 2.当有多个java服务时,会有多个jvm,也会有多个锁监视器,这样没办法使得多个jvm之间的线程互斥,所以无法使用jvm内部的锁监视器,也就是s

    2023年04月25日
    浏览(46)
  • 分布式锁实现(mysql,以及redis)以及分布式的概念

    我旁边的一位老哥跟我说,你知道分布式是是用来干什么的嘛?一句话给我干懵了,我能隐含知道,大概是用来做分压处理的,并增加系统稳定性的。但是具体如何,我却道不出个1,2,3。现在就将这些做一个详细的总结。至少以后碰到面试官可以说上个123。 那么就正式进入

    2024年01月21日
    浏览(62)
  • 分布式系统面试全集通第一篇(dubbo+redis+zookeeper----分布式+CAP+BASE+分布式事务+分布式锁)

    什么是分布式 一个系统各组件分别部署在不同服务器。彼此通过网络通信和协调的系统。 也可以指多个不同组件分布在网络上互相协作,比如说电商网站 也可以一个组件的多个副本组成集群,互相协作如同一个组件,比如数据存储服务中为了数据不丢失而采取的多个服务备

    2024年04月11日
    浏览(50)
  • Redis 分布式缓存

    单点 Redis 的问题及解决 数据丢失:实现Redis数据持久化 并发能力:搭建主从集群,实现读写分离 存储能力:搭建分片集群,利用插槽机制实现动态扩容 故障恢复能力:利用哨兵机制,实现健康检测和自动恢复 RDB RDB全称Redis Database Backup file (Redis数据备份文件),也被叫做

    2024年02月10日
    浏览(52)
  • Redis分布式缓存

    -- 基于Redis集群解决单机Redis存在的问题 单机的Redis存在四大问题: Redis有两种持久化方案: RDB持久化 AOF持久化        RDB全称Redis Database Backup file(Redis数据备份文件),也被叫做 Redis数据快照 。简单来说就是把 内存中的所有数据都记录到磁盘 中。当Redis实例故障重启后,

    2024年02月12日
    浏览(52)
  • 分布式Redis详解

    Redis我们最近学习必备工具之一了, 接下来我们将讲解Redis的简单应用 ,以及相关原理 第一种: 在Linux上安装Redis yum -y install redis 启动Redis redis-server /etc/redis.conf 链接Redis redis-cli 出现这个说明链接成功了 简单的读和取数据 设置远程链接 将redis的配置文件下载到本地: 配置文件是

    2024年02月13日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包