企业官网建设流程全解析

微网站开发需要几个人,社交类电商平台,垦利网站建设,WordPress状态栏替换#x1f345; 点击文末小卡片#xff0c;免费获取软件测试全套资料#xff0c;资料在手#xff0c;涨薪更快 最近在做大项目性能测试的工作#xff0c;又发现了几个比较有意思的Bug#xff0c;本期分享其中的一个#xff0c;涉及Redis在并发场景下的应用。有意思的是…点击文末小卡片免费获取软件测试全套资料资料在手涨薪更快最近在做大项目性能测试的工作又发现了几个比较有意思的Bug本期分享其中的一个涉及Redis在并发场景下的应用。有意思的是这个Bug因为没有代码语法错误并发量少的情况下下游的监控也不会出现报警所以光靠功能测试是没有办法发现只能通过压测性能测试或者下游的监控报警才能发现我们先来看一段Go语言实现的代码没有代码基础也没关系先解释一下这段代码的意思就是先获取Get Redis Key 的值这个值只有true 或者 false 两种情况 如果是true 则直接返回不执行后续代码逻辑如果是 false 则 先设置Set Redis Key 的值为true再执行后续代码逻辑var flag false // 获取锁 redisCache : redis.NewCache() if err : redisCache.Get(SetUserSizeRedisKey, flag); err ! nil { ctx.WarningF(request redis to get _user_size_flag fail, err: %s, err) } // flag true说明已经有实例在请求了, 直接返回 // 否则 设置redis锁 if flag { ctx.Notice(request_user_size_flag is true, return) return } else { ctx.Notice(request_user_size_flag is false, request im to flush room user size) if err : redisCache.Set(SetUserSizeRedisKey, true, ScriptMaxRunTime); err ! nil { ctx.WarningF(set request_user_size_flag to redis fail, err: %s, err) } } // ... 后续具体的业务代码逻辑可忽略然而单纯利用从Redis 当中获取一个Bool值以此来充当互斥锁这种实现方案在同一时刻只有一个用户请求能满足需求但是在并发场景会出现无法锁住的情况如下图在初始条件下即Redis Key 还从来没有被Set时Key不存在时当3个用户同时从Redis 读取到的值均为False 就有3个用户同时去Set Redis Key并且走到后续的代码逻辑所以并发场景下“锁”失效了锁失效了有什么影响继续给出完整代码逻辑这段代码其实是定时任务的一部分在执行期间会请求下游服务获得相关数据在并发场景下“锁”失效了会导致下游的服务压力上涨假设下游只能抗50QPS现在QPS 已经到5000了严重情况下还会出现IO打满CPU和内存打满服务宕机等风险package main importtime var ( CrontabTime 20// 每20s执行一次脚本 ScriptMaxRunTime 150// 脚本最长运行时间150s ) func SetUserSize(ctx *gin.Context) { var flag false // 获取锁 redisCache : redis.NewCache() if err : redisCache.Get(SetUserSizeRedisKey, flag); err ! nil { ctx.WarningF(request redis to get _user_size_flag fail, err: %s, err) } // flag true说明已经有实例在请求了, 直接返回 // 否则 设置redis锁 if flag { ctx.Notice(request_user_size_flag is true, return) return } else { ctx.Notice(request_user_size_flag is false, request im to flush room user size) if err : redisCache.Set(SetUserSizeRedisKey, true, ScriptMaxRunTime); err ! nil { ctx.WarningF(set request_user_size_flag to redis fail, err: %s, err) } } // 以下是定时任务的具体逻辑 ctx.NoticeF(run time start:%d, util.GetMilliSecond()) // 执行定时任务前前置获取相关必要信息 info, err : GetInfo() if err ! nil { ctx.WarningF(request info fail, error: %s, err) // 获取信息失败提前释放锁 _ redisCache.Del(SetUserSizeRedisKey) return } ctx.NoticeF(run time start req im:%d, util.GetMilliSecond()) // 定时任务具体逻辑 for _, people : range info { //请求下游 // ... } ctx.NoticeF(run time end:%d, util.GetMilliSecond()) }如何解决请求下游数量超限这个问题呢有两种解法第一种是在请求下游前增加判断当前QPS。第二种是使用Redis 分布式锁setnx限定QPS先看第一种方案是在请求下游前判断是否超过最大的QPS如何获取QPS呢QPS是在做性能测试时我们常用的性能指标指每秒的查询数量用来衡量系统每秒处理的请求数量那么要获取QPS自然要获得当前的秒数如果是在同一秒请求我们用当前秒数作为Redis Key 值初始为0同一秒内每有一次请求就把Redis 的值加1这样就拿到了QPS见代码当中的GetLimitRequest方法now : util.GetSecond() nowMilli : util.GetMilliSecond() res : GetLimitRequest(ctx, now) // GetLimitRequest 获取当前qps func GetLimitRequest(now int64) int { key : fmt.Sprintf(limit_key_request_service_%v, now) redisCache : redis.NewCache(ctx) res, _ : redisCache.Incr(key) if res 0 { go redisCache.Expire(key, 60) } return res }那QPS超出限额了怎么办得计算到下1秒还有多少时间要精确计算的话我们只能可以获取比秒更小的单位-毫秒进行计算分别获取下1s的时间毫秒为单位以及当前这1s的时间同样毫秒为单位两者相减这样就知道到下1秒还差多少毫秒对应下面代码的变量gap让系统sleep gap对应毫秒数这样就可以使得请求的维持在最大的QPS范围内完整的代码片段如下package main importtime var ( CrontabTime 20// 每20s执行一次脚本 ScriptMaxRunTime 150// 脚本最长运行时间150s ) func SetUserSize() { var flag false // 获取锁 redisCache : redis.NewCache() if err : redisCache.Get(SetUserSizeRedisKey, flag); err ! nil { ctx.WarningF(request redis to get _user_size_flag fail, err: %s, err) } // flag true说明已经有实例在请求了, 直接返回 // 否则 设置redis锁 if flag { ctx.Notice(request_user_size_flag is true, return) return } else { ctx.Notice(request_user_size_flag is false, request im to flush room user size) if err : redisCache.Set(SetUserSizeRedisKey, true, ScriptMaxRunTime); err ! nil { ctx.WarningF(set request_user_size_flag to redis fail, err: %s, err) } } ctx.NoticeF(run time start:%d, util.GetMilliSecond()) // 执行定时任务前前置获取相关必要信息 info, err : GetInfo() if err ! nil { ctx.WarningF(request info fail, error: %s, err) // 获取信息失败提前释放锁 _ redisCache.Del(SetUserSizeRedisKey) return } ctx.NoticeF(run time start req im:%d, util.GetMilliSecond()) // 定时任务具体逻辑可忽略 for _, people : range info { now : util.GetSecond() nowMilli : util.GetMilliSecond() res : GetLimitRequest(ctx, now) if res MaxQps { gap : (now1)*1000 - nowMilli if gap 0 { time.Sleep(time.Duration(gap) * time.Millisecond) // 在sleep 期间 不再请求下游 } } elseif res 0 { //异常 time.Sleep(time.Duration(40) * time.Millisecond) } } // 请求下游具体代码逻辑可忽略 // ... // ... // 定时任务执行完毕主动释放锁 _ redisCache.Del(SetUserSizeRedisKey) ctx.NoticeF(run time end:%d, util.GetMilliSecond()) } / GetLimitRequest 获取当前qps func GetLimitRequest(now int64) int { key : fmt.Sprintf(limit_key_request_service_%v, now) redisCache : redis.NewCache(ctx) res, _ : redisCache.Incr(key) if res 0 { go redisCache.Expire(key, 60) } return res }使用Redis分布式锁setnx是Redis的一个命令它代表Set if Not eXists。这个命令尝试在Redis中设置一个键值对但仅当指定的键不存在时才会成功。如果键已经存在setnx操作将失败我们可以使用setnx命令来创建Redis分布式锁分布式锁是一种机制用于确保在分布式系统中的多个节点或线程不会同时访问或修改共享资源以避免竞态条件race conditions分布式锁的主要目的是确保在分布式系统中只有一个客户端或线程能够成功获得锁以执行关键任务而其他客户端必须等待setnx的使用方式是客户端通常会使用setnx命令尝试创建一个带有唯一标识的锁然后在锁上设置一个过期时间以防止锁被永久占用。当客户端不再需要锁时可以使用del命令来释放锁对于上面的并发问题我们还可以使用SetNX来解决func SetUserSize(ctx *gin.Context) { ExpireTime: int64(3) client : redis.NewCache(ctx) key : fmt.Sprintf(set_locker_%s, param_ex) res, err : client.SetNX(ctx, key, time.Now().Unix(), ExpireTime) //创建redis 分布式锁ExpireTime过期时间为3秒 if err ! nil { ctx.WarningF(SetQuestionStatus get lock fail, err: %v, err) errno.ErrRet(ctx, errno.ErrCallCacheFail) return } if res ! true { errno.ErrRet(ctx, errno.ErrSetInfo) return } defer client.Del(ctx, key) //执行完删除Redis分布式锁让其他线程能正常获取锁避免永久等待 //... 执行后续逻辑 }用一张图片再来对比一下两种实现方案的区别使用Redis分布式锁能帮助解决高并发下互斥任务的问题但需要注意设置过期时间避免永久锁住资源后续我会继续分享压测中发现的性能问题以及排查、调优实战解决方案最后感谢每一个认真阅读我文章的人礼尚往来总是要有的虽然不是什么很值钱的东西如果你用得到的话可以直接拿走这些资料对于做【软件测试】的朋友来说应该是最全面最完整的备战仓库这个仓库也陪伴我走过了最艰难的路程希望也能帮助到你凡事要趁早特别是技术行业一定要提升技术功底。