本文详细描述了腾讯自选股业务中如何实现：目前手上可用的资源仅剩一台16核剩余4-8G内存的机器，单点完成了在1个小时内千万级别feed流数据flush操作（主要包括：读数据，计算综合得分，淘汰低分数据，并更新最新得分，回写缓存和数据库）
腾讯自选股App在增加了综合得分序的Feed流排序方式：需要每天把（将近1000W数据量）的feed流信息进行算分计算更新后回写到数据层。目前手上的批跑物理机器是16核（因为混部，无法独享CPU），同时剩下可用内存仅4G。显而易见的是：我们可以申请机器，多机部署，分片计算或者通过现有的大数据平台Hadoop进行运算都看似可以解决问题。但是由于更新feed流的操作需要依赖下游服务（这里暂且叫A，后续文中提到下游服务均可称A服务），而下游的服务A-Server本身是个DB强绑定的关系，也就说明了下游的服务瓶颈在于DB的QPS，这也导致了即便我本身的服务多机部署，分片处理，下游服务的短板导致不可行。而针对方案二通过大数据平台完成的话，也就是需要推荐大数据的部门协助处理，显然这个是需要排期处理，而时间上也是不可预估。
既然如此，那就借用，朱光潜老先生的一篇文章《朝抵抗力最大的路径走》。我本人相信通过合理的资源调度以及更低的成本可以克服眼前的困难，实现最终的需求效果。当然优化过程中并不是一帆风顺，当然经过两周左右的优化迭代，也终于实现了。
整个flush的业务流程大致如下：
- 读取DB获取目前所有的feed类别（约2-3w的数据）
- 通过类别读取Cache每一个类别下的feed流元素的索引（约1000-10w的数据）
- 通过每一个信息的索引查询feed流所对应的基础数据信息（需要查约3-4张表）
- 计算每一个feed元数据的得分信息（1000w的数据量），过程中需要淘汰一部分，调用服务A-Server删除当前的索引
- 根据权重计算每一个feed的元素的信息，调用下游服务A-Server，update索引分值
主要业务流程图具体如下
针对上述的业务逻辑，设计出了最初方案
- 查询DB或者本地缓存获取索引feed流中的现有全集类别
- foreach类别集合Collection，查询目前所以的类别下的feed数据流集合并存储到Map中，其中key是类别，value是类别对应的数组集合(key:category,value:colletion)
- foreach上述获取的Map并发起goruntine查询每一条信息流元素对应的基本信息，并通过粗排来淘汰需要淘汰的元素（考虑到下游的并发和DB的负载问题，每查询一批，sleep一段时间），把最终符合要求的元素存储到map等待后续更新得分，并刷入缓存和DB
- foreach上述粗排后的Map，最终并发起goruntine调用下游A-server，更新feed流的索引得分
方案图如下
最初方案缺陷
尽管考虑到将近1000W的数据虽然在处理过程中，内存会是问题，于是在使用后的集合或者Map都会及时清空
Map=nil []string=nil // 清空已使用的内容
runtime.GC() // 发出GC的请求，希望发起GC
但是问题还是出现了
内存跑满（由于机器总内存18G，所以基本是内存直接跑满了）
Cpu也基本瞬间跑满
堆栈中的异常
compress@v1.12.2/zstd/blockdec.go:215 +0x149
created by github.com/klauspost/compress/zstd.newBlockDec
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:118 +0x166
goroutine 61 [chan receive, 438 minutes]:
github.com/klauspost/compress/zstd.(*blockDec).startDecoder(0xc006c1d6c0)
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:215 +0x149
created by github.com/klauspost/compress/zstd.newBlockDec
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:118 +0x166
goroutine 62 [chan receive, 438 minutes]:
github.com/klauspost/compress/zstd.(*blockDec).startDecoder(0xc006c1d790)
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:215 +0x149
created by github.com/klauspost/compress/zstd.newBlockDec
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:118 +0x166
goroutine 63 [chan receive, 438 minutes]:
github.com/klauspost/compress/zstd.(*blockDec).startDecoder(0xc006c1d860)
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:215 +0x149
created by github.com/klauspost/compress/zstd.newBlockDec
/root/go/pkg/mod/github.com/klauspost/compress@v1.12.2/zstd/blockdec.go:118 +0x166s
因为堆栈给的信息不多，但是从机器上看基本是goruntine开启的太多，并发量太大，同时大量的数据同时加载到内存，导致了机器的内存和Cpu的负载过高
- 自己实现一套协程池
- 预分配一个内存块，维持一个对象池
对象池具体改进点如下
协程池
实现比较简单，这里就直接上代码
// 协程池对象
type PoolBuilder struct {
workerNum int // Worker 线程数量
DelJobChan *chan string // 缓冲队列
}
// 创建一个协程池
func (pool *PoolBuilder) listenAdd(num int) {
for i := 0; i < num; i++ {
go func(i int) { //
addWorker(pool.AddJobChan)
}(i)
}
}
// 任务写入队列
func (pool *PoolBuilder) InsertAddChannel(id string, score int64) {
log.Infof("send value to add channel,%s", id)
pool.AddJobChan.In <- &AddChannel{
id: id,
score: score,
}
}
优化后的方案缺陷：
- 内存和Cpu的负载相对降下来了，但是由于下游服务A-Server是对DB的强依赖的类型，所以突然的高并发，DB的瓶颈成了A-Server的服务瓶颈
- 如果并发量降下来，但是6个小时内完成1000w的数据读库，业务计算，算法排序以及删除和更新每一条数据的得分，显然不够
全量的数据计算，并发高，下游服务，下游存储资源扛不；相对并发不高的情况，数据计算不完。与组内小伙伴商量，可以采用大数据平台计算不失一种好的办法。看似最优解，但是大数据平台接入，以及推动大数据平台的开发也是需要走排期等流程。
经过了两周的专研和思考，我最终从：hadoop的mapreduce分而治的思想、vert.x的全异步链路（本人超级喜欢的一个框架，使用后，根本不想写同步代码了 ）以及Linux的内核调度机制的三种优秀的设计中借鉴了一些思路，最终完成了40分钟跑千万级别的数据优化！
1、Hadoop的mapreduce 分而治的思想
把任务拆分成若干分，然后分配给一个woker
让每个worker处理手中的任务，并把处理后的子任务汇集到一个woker-Awoker-A负责把所以的子任务结果，汇总处理，并返回
启发
我可以把每一个类别分配给一个协程处理，而每一个协程只负责每一个类别下的所以数据，这样协程的数量也就是类别的数据，这样进一步节省了协程数量，但是由于merge的结果在最终一步，这样的话内存就需要存储处理后全量数据，这一点与目前的内存有限不符合，所以这里借鉴了把任务分发的思想
2、Linux的内核调度机制（非epoll）
在Linux的中内核调度中，我们知道非epoll的模式中，无论是poll和select的时候，都会有一个select来负责后续的任务调用和分配，用官方的描述就是：select 轮训设置或检查存放fd标志位的数据结构进行下一步处理。如果满足状态，就会扭转到下一个步，唤起相应的进程函数调用。
启发
这里可以参考select这个负责任的角色，当然改进的地方是我可以增加多个协程来并发查询所以类别，并进行分发类别处理，这样话，下游的协程池就可以尽可能的在完成一次调度后，马上进行下一次调度（因为分配任务的协程多了），而不会进入调度空闲的状态。
这里就直接使用网上的一张图：
3、vert.x全异步链路
我将这个vert.x 标红了，可以看到这里vert.x给我的启发是最关键也是最大的。上述问题，我反复思考，我发现，其实我如果突然的高并发，必然导致了下游的服务负载过高从而导致DB和下游服务扛不住。如果我能平滑的并发，而不是从某个时间点起，并发操作，也许就能解决这个问题！
并发代码我们写的多，但也许我们大家写的只是并发而不是真正的异步，因为我们在开始或者函数汇总的结果初我们都会使用阻塞，当然我也是有短时间没有写全异步的代码了，所以思想固化了，具体案例如下分析：
这种在主线中启动并发或者异步的处理，最终还是需要在主线程中使用wait来阻塞等等所以线程的结果处理完毕，这样看似提高的吞吐量，但是由于需要对并发线程或协程的结果进行汇总计算，这样就注定要把大量的结果集合存储到内存，然后进行后续的操作。这样的异步更像一种伪装异步。
而在vert.x 中是将上下游的数据通信都是用了callback的方式处理，而正是这样，这个框架的做到了全链路的异步逻辑。这里我们看看这个框架的核心思想：
Vertx完成采用另一个机制，用一个线程来接受请求（也可以是几个，注意是几个，不是几百个），而把这个真正要执行的任务委托给另外一个线程来执行，从而不会堵塞当前线程
另外在Vert.x中的调度模型也正是使用了Linux的epoll的事件驱动的机制，大致如下
整体来看vert.x的做到了：
1.非阻塞处理请求，异步执行阻塞程序，保证了请求处理的高效性。
2.使用Event Bus事件总线来进行通讯，可以轻松编写出分布式、松耦合、高扩展性的程序
这里可以展示一下Vert.x的异步代码
public class Server extends AbstractVerticle {
public void start() {
vertx.createHttpServer().requestHandler(req -> {
req.response()
.putHeader("content-type", "text/plain")
.end("Hello from Vert.x!");
}).listen(8080);
}
}
对异步代码有兴趣的小伙伴一定要看看：https://vertx.io/
借鉴了上述优秀的思想，我对自己的服务做了以下改进：
1、我构造了4个协程池，分别是查询类别category、查询DB基本信息、根据算法计算综合得分、和数据更新回写；
2、从主协程开始，不做任何阻塞，查询类别的协程协程池，每查询一个类别，结果直接丢到channelA（不阻塞然后继续擦下类别）
3、查询DB的协程，监听channelA，当发现有数据的时候，查询DB信息，并将结果丢到channelB（同上不做任何阻塞，继续查询下一条数据的结果集合）
4、帖子得分协程池读取channelB的数据，然后根据算法计算处理帖子的得分，并将结果集合丢到channelC（同样不做任何阻塞，继续计算下一次的得分数据）
5、而数据回写的协程负责调用下游服务A-Server，处理后完，打log，标记处理的偏移量（由于没有阻塞，需要跟着最终所以数据是否处理完成）
业务架构设计如下：
1、协程数6w->100！ ，这里协程数从6w降到了100个协程就Cover住了整个项目
2、内存使用情况，从基本跑满到仅仅使用1-2G的正常内存
3、CPU的使用460% 的使用率直接降到65%
4、计算数据量1000w的时间6个小时并发算不完到46分钟计算完成！
2022年一个新的开始，没想到自己的坚持看到了效果，自选股的业务中也因此可以接入综合得分序列的feed流，我相信这个是一个好的开始，在这个基础上，我们可以根据个人画像做更多的智能推荐，期间大伙的建议更多是借用大数据平台计算，而实际的推进和排期让我更愿意用自己的方式以最低的成本最优的结构去优化完成，当然这次很幸运，自己的努力实现了！