一次 Redis 作为中间层缓冲数据读写的经历

公司有个需求，收集机器数据。使用的是 Telegraf，启动一个 Client 接收 Telegraf 日志，并批量上传到服务器 Server，服务器通过 Redis 缓存数据，之后 Consumer 负责读取批量写入 InfluxDB。

第一个让人为难的问题，读取写入顺序问题——软件工程没有银弹。

其实我一直想用 Kafka 实现这个需求，但一开始只有 Redis，那就先用 Redis List 结构实现，但有一个很严重的问题，Redis List 不支持消息队列的 ACK 模式。仔细一下，那就自己简单模拟下，读数据，删数据。关键是 Redis 批量读数据、写数据还不是原子操作，就有两条路可走：

1. 读 Redis，写入 InfluxDB，再删 Redis。防止写入失败，可以重试，InfluxDB 重复写入会覆盖。但这有个问题，读出来的数据一致有问题咋办？死循环。并且，写入 InfluxDB 的时间还要加锁，防止别人读到
2. 读 Redis，删 Redis，写入 InfluxDB。防止发生错误，读取死循环。但要是写着写着 InfluxDB 掉线，数据就丢了。

后来抉择下，选择了 2。各有利弊，只是在公司层面上 2 其实很合适一点，InfluxDB 用的挺稳定，但数据量很大，加锁等待写入，不能接受。

然后是一个 bug，线上偶尔丢几条数据。

Server 是这样保存 Client 数据的：

RPUSH telegraf dataA dataB ...

Consumer 是这么消费数据的：

setnx lock serverA
size = llen telegraf
# fetchCount 一次取得的数目
datas = lrange telegraf 0 fetchCount -1 
# 保留后面的数据
ltrim fetchCount size
del lock
# 处理数据
...

各位可以先看看有什么问题。

想着多个消费者同时消费，加个锁消费，所有人读到的数据不会重复，并发一个一个排队。Consumer 这边并发读取确实没有问题，但是我忘记了一件事，还有数据在源源不断的被 Server Push 进去，导致 size 一直在增加，而我截取的 size 是之前的 telegraf 的长度。

我特别郁闷，自测没问题，因为并发量不大，很少有情况会一边 PUSH，一边写入。并且为什么不是丢大量数据？因为 Telegraf 只在整分钟采集数据疯狂上传，基本上只有在整分钟那会才会触发。

后来我才知道 LTRIM 支持负数，表示从后向前的坐标，size 变成 -1 就行了。

最后遇到一个问题就更难以置信了。

我一开始批量读取 2000 个写入，能及时写入。之后加了很多机器，心想着变成 10000 个写入，不就完事了。

可惜我还是太年轻，变成 10000 个，反而积压了。慌得不行，赶紧回退。

写入的伪代码是这样：

data = int[10000]
rs = influxObject[10000]
for data in datas 
   parse data(json string) to object
   convert object to influx object
   rs += influxObject
write rs to influxDB

看得出来原因吗？代码中执行了 JSON 转换，再转换成 InfluxDB 对象的操作，取的越多，转换越慢。写入延迟就更长。所以并不是读取的越多消费的越快。这里需要并发去处理数据，这有两种做法：

1. 多起几个 Consumer，数量设置小一些，进程并发
2. 处理数据的时候，多起几个线程、协程

在不断地折磨之中，业务最终稳定了，性能也还不错。