之所以有这篇文章，是因为最近一段时间在帮助同事做性能优。目前优化工作大体上已经完成，现在回过头来才发现这段时间是对以往“八股文”在真正开发过程中的实际应用。包括但不限于：接口超时限制、预防缓存击穿、数据库读写分离，Redis管道操作，服务日志梳理等等。每部分似乎都是司空见惯，但串联起来，是一套较为完整的系统性的应用，还是非常值得记录下来的。
我们的产品是面向全球的Web应用，服务器部署于AWS的EKS集群(AWS的K8S产品)之上，目前采用的是微服务的形式，但不是服务网格(Service mesh)的方式，一些基础设施是公用的，例如Redis集群、数据库实例、消息队列等等，一部分原因是因为最开始的技术架构设计，一部分是需要考虑到成本，我司属于创业公司，是典型的预算敏感型用户。
此文章介绍的内容是产品下的其中一个模块，在决策层面也是接下来的宣发和推广的重点。就此模块而言，我们的角色是数据的接入方，我们的供应商提供了基于Webscoket的数据流订阅机制，我方需要订阅相应的数据流，对数据进行清洗整理，再做一定的聚合，之后为客户端提供HTTP接口。在与供应商的沟通中我们了解到，他们数据推送的峰值可能达到7GB/min，这是一个非常恐怖的数据，因此数据消费处理过程将面临不小的挑战。
前面说到，供应商的数据推送量会比较大，显然数据消费处理过程是首当其冲的优化点。经过对需求和数据特点的分析评估，我们对数据进行分级，分为热数据和持久化数据。
其中，热数据的特点是：1.高频的访问和更新，2.数据属于元数据可以直接存储。
持久化数据的特点是：1.低频写入，2.中低频访问，3.结构化的。
鉴于上面数据的不同特点，我们决定将热数据存放到缓存也就是Redis中，将持久化数据存放到数据库中。
少啰嗦，先看东西
为了阅读更加清晰，非关键代码已省略。
...
func feedStream() {
...
var messages = make(chan string, 1024*5)
go readFeedMessages(connect, messages)
wd := &sync.WaitGroup{}
for i := 1; i <= 100; i++ {
wd.Add(1)
go feedHandler(wd, messages)
}
wd.Wait()
}
...
这是改造前的数据消费代码段。从代码中你可以看到，首先为消费创建了一个通道，长度为1024*5，然后开启了100个协程对这个通道进行并发读取。
我们知道go的通道为了实现并发安全，其内部实现是有锁的。这里，我的想法是降低锁的影响范围，也就是常提到锁粒度。因此，我把代码改造为为各自的协程分配各自的通道。
...
//初始化通道
var (
maxMessageChannelNum = 200
messageFeedChannels []chan string
)
...
for i := 0; i < maxMessageChannelNum; i++ {
messageFeedChannels = append(messageFeedChannels, make(chan string, 1024))
}
...
// 读取消息
func readFeedMessages(connect *websocket.Conn, messageChannels []chan string) {
var processStep int
...
for {
_, r, err := connect.NextReader()
...
feedBuf, err := io.ReadAll(r)
...
//将消息均衡的分配到不同的通道中
if len(feedBuf) > 0 {
messageChannels[processStep] <- sailUtils.BytesToStr(feedBuf)
processStep++
if processStep == maxMessageChannelNum {
processStep = 0
}
}
}
}
以上代码已经忽略了错误处理，通道关闭等逻辑。
细心的你可能已经发现了这段代码：
sailUtils.BytesToStr(feedBuf)
根据方法名称你也猜到了，它的作用就是将byte数组转换为字符串。那么为什么不用内置方法string()呢？这里有一个小小的前提，那就是供应商发送单条数据都会比较大。使用内置方法性能相对较低。(我们的golang版本是go1.19)。相应的benchmark请参考鸟窝大佬的这篇文章：四种字符串和bytes互相转换方式的性能比较
我们使用的缓存组件是Redis，基于它高性能和可靠的表现，本没有什么问题。但是我们还是得为领导们提前谋划(提前预判领导们的预判)：成本。
因此，我们在将数据存入Redis之前，做了对大Key一个额外的操作：压缩。
之所以这么做，我们分析如下：
1.压缩的性能消耗点在CPU上，通过对服务器的性能监控，发现当前CPU的使用率比较低，可以匀一部分出来。
2.即便日后服务器的CPU性能不够了，扩容服务器的成本也远低于Redis的成本。
3.在可观的压缩率面前，除了节省缓存空间，还能额外的节省网络带宽，降低延迟。
在进行压缩操作之后，Redis的空间消耗相比之前减少了4/5。
优化前(小峰值)：
优化后(加上了过期清理)：
经过对代码的走读，发现许多的操作是可以合并的，因此，我们的可合并的操作合并之后进行管道操作。
...
pipe := sail.GetRedisCluster().Pipeline()
var cmds = make([]*rdb.SliceCmd, 0, len(marketIds))
for _, marketId := range marketIds {
redisKey := fmt.Sprintf("%s:eventId:%d:markets", config.GetGlobalConfig().RedisPrefix, eventId)
cmd := pipe.HMGet(ctx, redisKey, fmt.Sprintf("%d", marketId))
cmds = append(cmds, cmd)
}
...
_, err := pipe.Exec(ctx)
...
Redis的管道操作可以减少命令发送和响应的往返时间、提高吞吐量、减少命令执行的总延迟。在这里的高并发场景下，性能提升显著。
附上一张峰值下的Redis监控图：
Redis: 我现在强的可怕！（狗头+破音
说完了数据消费，接下来就是将清洗好的数据以HTTP接口的形式提供给前端。
这里附上一张优化前的接口响应截图：
可以看到，整个请求耗费了51.07s，这相对来说是比较极限的情况，但此接口绝大部分时候耗时都在30s上下，这显然是不可接受的。
在走读了代码之后，发现并没有对接口的响应时间进行约束控制，因此，首先将这项加上。
...
func MatchInfo(c *gin.Context) {
var (
form request.MatchInfoRequest
resp response.MatchInfoResponse
loggerSvc = c.MustGet("logger").(*zap.Logger)
ctx, cancel = context.WithTimeout(context.Background(), time.Second*10) //目前取保守值为10s
)
defer cancel()
...
}
之后的操作，无论是Redis还是数据库，都会使用这个context。
result, err := sail.GetRedisCluster().Get(ctx, redisKey).Result()
err := sail.GetDBW().WithContext(ctx).First(&record, "eventId = ?", form.EventId).Error
加上之后，立马就“见效”了。
一定要加上接口超时约束限制是为了有效的防止服务雪崩。当出现大量挂起的请求时，整个服务的响应会迅速降低，内存也会疯涨。如果请求一直挂起得不到响应，那么前面说的情况就会迅速恶化，很快整个服务就会宕掉。
上文中已经讲过了，这里不再赘述。
在给客户端提供的接口里面，绝大部分都是查询操作，因此，我这里将查询操作打到读实例上面。使用go-sail提供的语法糖：
err := sail.GetDBR().WithContext(ctx).First(&record, "eventId = ?", form.EventId).Error
另外，部分数据表属于大表，字段较多，个别字段还是text类型。因此，针对一些查询操作，只提取用到的字段，而不是全部：
err := sail.GetDBR().WithContext(ctx).
Select("id", "name").
First(&record, "eventId = ?", form.EventId).Error
对于一些高频访问的热点数据，没有必要实时读取整合，在保证数据时效性的范围内做缓存，比如20s。为了防止缓存击穿，需要对操作进行加锁处理。
...
lockerKey := fmt.Sprintf("%s:MatchInfoNX:%d", config.GetGlobalConfig().RedisPrefix, form.EventId)
redisKey := fmt.Sprintf("%s:MatchInfo:{%d}", config.GetGlobalConfig().RedisPrefix, form.EventId)
result, err := sail.GetRedisCluster().Get(ctx, redisKey).Result()
if len(result) == 0 || err != nil {
var locked bool
//当缓存中的数据不存在时，争抢锁，只放行一个请求去数据库查询，自旋10次
for i := 0; i < 10; i++ {
locked = utils.RedisTryLock(lockerKey)
if !locked {
time.Sleep(time.Millisecond * 200)
continue
}
break
}
if locked {
defer utils.RedisUnlock(lockerKey)
}
...
} else {
...
resp.Data = matchData
sail.Response(c).Builder(constants.ErrNone, resp).Send()
return
}
...
你可能会问，那缓存穿透问题怎么解决，是使用了常规的布隆过滤器吗？并没有，这里的业务逻辑相对简单明了，请求的参数必须要在我们提供的名单之中，否则会被拒绝。
对于一些不必要的查询，该去就要去掉，简而言之，相同情况下，如果能读取一次数据库能完成，那就不要读取两次。（请不要杠查询复杂度有没有增加这类问题，说的是相同情况下🙂
对于一些可以优化的查询方式，也可以考虑用更加合理的方式处理，比如，走读代码发现一个地方对一个大Key有不必要的读取：
result, err := sail.GetRedisCluster().HGetAll(ctx, redisKey).Result()
这个Key是一个有几百KB内容的哈希，而读取它只为了检查它有多少个键，没有必要将内容都读取出来。并且，这个操作还是在循环里面的。因此，我们将这里的操作更改为Redis提供的HLen方法：
cnt, err := sail.GetRedisCluster().HLen(ctx, redisKey).Result()
日志是一个全局性的内容，放到这里正好可以和链路追踪一起讲。我们的日志是使用的uber/zap库。在go-sail框架中，我对zap进行了一定的封装，实现了对链路追踪的支持，让它可以方便无感的导出到ELK中。更多内容可移步至：
打印日志也是会消耗性能的，特别是在量大之后会发生质的变化。走读代码之后，发现对日志的使用有许多不合理的地方。得到运维同事的反馈，近来的ELK日志量非常的大：
因此，首先对日志进行梳理。除了zap本身的日志级别分级之外，我对日志类别做了重新的划分：
1.控制台日志
2.zap日志
对于调试性的日志，统一输出到服务控制台，并且加上了debug模式的约束，当调试结束，关闭debug之后，就不再输出。对于需要持久化的日志，由zap统一输出到ELK。其中又将zap日志拆分为计划任务日志和接口日志，让后续查询更加清晰。另外，使用请求上下文中的logger实例，这样有助于链路追踪。最后，清理掉冗余的日志。
控制台日志：
if config.GetGlobalConfig().Debug {
fmt.Println("feedChanLength:", len(messagesChan))
}
请求上下文中的logger实例：
func MatchInfo(c *gin.Context) {
var (
...
loggerSvc = c.MustGet("logger").(*zap.Logger)
...
)
...
loggerSvc.Error("...", zap.String("err", err.Error()))
}
梳理完之后，体积有了显著的改善：
结合上述一系列的优化，接口响应延时得到了明显的改善：
常规情况下，接口返回在毫秒级别。极端情况下（数据量大且未命中缓存），控制在3s范围内。
上文中有一个截图里面提到，如果更改了Redis实例类型需要批量替换部分代码。这一点确实“不智能”。于是我去翻看了go-redis库的部分源代码后，发现它的作者实际上早就提供了一个公共的抽象接口，用于隐藏不同连接的差异。因此，我对go-sail的redis操作进行了调整：commit
罗马不是一日修建而成的，这是一个循序渐进的过程，在这个过程中，需要时刻保持着一个最简单的想法，那就是“苍蝇再小也是肉”，你最终会发现集腋成裘，功法已成。
觉得博主写的不错？给他一个赞赏：
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。