近期发现,开发功能的时候发现了一个 mq 消费顺序错乱(历史遗留问题),导致业务异常的问题,看看我是如何解决的
首先,简单介绍一下情况:
线上 k8s 有多个 pod 会去消费 mq 中的消息,可是生产者发送的消息是期望一定要有序去消费,此时要表达的是,例如 生产者如果发送了 3 个通知消息,分别是
此处,若有 3 个 pod 的分别拿到了上述 3 条消息,但是自身实际消费完毕的顺序可能是 先完成了 3 消息对应的业务逻辑,再是 2 消息 的业务逻辑,最后是 1 消息的业务逻辑
那么这个时候,小 d 用户就没有绑定上 允许看视频类型的网站 这一条策略,自然 b组 和 a 组也没有绑定上这条策略,这就和我们预期的完全不一致了
当然,实际情况对于单条单条的消息处理基本不会出现这种偏差,但是在批量处理的时候,就会出现实际业务处理顺序与期望不一致的情况,那么就是妥妥的线上问题了(小 d 上网的时候想看视频,可是一直看不了,于是就疯狂投诉。。。)
对于这个问题如何解决呢?
我们知道,咱们使用 mq 的目的是为了做到去处理我们的异步逻辑,还能对流量进行削峰,服务间解耦
对于咱们的 A 服务,已经处理了关于添加用户的,添加组的逻辑,发送通知消息给到 B 服务的时候,B 服务自身的处理顺序,未按照既定的顺序真实按照顺序消费完毕,导致出现了业务问题
想法一
我们是期望 B 服务团队去添加批量接口,A 服务将需要通知的信息,排序好给到 B 服务,一个整包, B 服务的单个 pod 接收到这个大包,然后按照顺序处理消息即可,但是这个方式弊端比较明显
想法二
对于这一个业务,也不能去对整个架构大改,对于这些历史遗留问题,能少动就少动,兄弟们你们都懂的
于是便想出了使用 redis 分布式锁来处理,对于一个部署在 k8s 中服务的多个 pod 去抢占,谁先抢到锁,那么就谁消费 mq 中的消息,没有抢到锁的 pod ,那就过一会再抢
当然,对于其他类型的业务是没有影响的
如何去实现这个想法呢,我们可以模拟一下
这样来处理的话,我们就可以应对多个 pod 来消费同一类消息的时候,保证同时只有一个 pod 在处理 mq 中的消息了,当然如果正在处理消息的 pod 出现了异常,对于其他 pod ,最晚会在 40 秒之后拿到锁,对于大量的消息来说,这个还是可以容忍的
对应的代码逻辑如下:
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)
}
}
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
}
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
程序 b 日志如下:
程序 b 先是获取锁失败,过 30s 左右,程序 b 能正常获取到锁
关于源码可以查看地址:https://github.com/qingconglaixueit/my_redis_demo