埋点日志最终解决方案——Golang+Gin+Sarama VS Java+SpringWebFlux+ReactorKafka

之前我就写过几篇OpenResty+lua-kafka-client将埋点数据写入Kafka的文章，如下：

Lua将Nginx请求数据写入Kafka——埋点日志解决方案

python定时任务执行shell脚本切割Nginx日志-慎用

nginx+lua写入kafka报buffered messages send to kafka err: not found broker

关于OpenResty+doujiang24/lua-resty-kafka写入kafka故障转移模拟测试

以上一步一个坑，有些是自己能力不够踩的，有些是为了解决某个问题踩的，最后终于消停的一阵。但又出现新问题了，这次问题没那么紧急，但比较重要。按照一般的剧本，上面的坑都踩完，基本上也就不会怎么去改这个服务，但新的问题还是出现了，就是容器化部署基础镜像要升级，从原来的debian10升级成了debian11,当然这是大版本，小版本几乎没周都会升级，升级时也不会通知项目组测试，运维直接升。在debian10升级debian11的时候，出现了一个问题细思极恐，就是zlib升级其中一个方法签名变了，导致我们lua脚本报错了，我们发现了这个问题，由此引出来一个担忧：运维升级小版本的时候会不会升级到某个我们用到的运行库导致线上出问题。评估下来发现非常有可能，因为运维升级镜像是基于一个镜像扫描软件，这个软件经常会扫描出诸如openssl这种组件的问题，要求运维在一个月内升级完成。这就很有可能影响到我们。并且，我们在升级kafka server的时候发现doujiang24出品的kafka-client很难像社区一样保持活跃更新，支持一些kafka新特性，并且有问题也难以求助（虽然上次得到他本人的回复，但一个人总比一群人回复问题滞后一些）。

解决思路

思路1

运维升级的时候通知到对操作系统组件敏感的我们，由我们评估是否需要跳过本次升级。这个比较难判断，因为我们项目组也无法精确的判断哪些组件一定会影响到我们，考虑不使用

思路2

将底层可能影响我们的组件进行后置，比如gzip和aes，放在kafka后面的flink去做，而不是在Nginx这里就处理掉。这个思路能避免底层升级带来的大部分影响，但是kafka驱动升级问题无法避免，考虑不使用

思路3

我们还有其它服务都是用Java做的，正式因为有JVM这一层的存在，我们才不怕操作系统的升级，是不是可以用Java实现，从而避免此问题。这个思路能解决上面的担忧，但是性能需要做测试，即使用NIO，想要达到目前的TPS，还是需要一定资源的，因为OpenResty和Java，达到同样的TPS，内存使用量差距还是很大的。这个思路保留，做进一步测试。

思路4

可不可以保持低资源高性能，又用一个中间层屏蔽操作系统组件升级带来的影响呢？这时我想到了golang。这个思路保留，做进一步测试。

思路5

这个项目本来的架构式OpenResty+Apache Flumed，是不是可以还原到这个架构，把OpenResty中的组件后置到Flumed中？这个也被否决的，原因有以下几点：

如果把OpenResty和Flumed部署到同一个容器中，因为公司标准的监控只能监控其中一个进程，如果某个进程挂了，可能无法监测到，这个问题在之前遇到过，一个容器内起了一个OpenResty和5个Flumed进程，其中某个Flumed进程挂了好久才知道
如果OpenResty和Flumed分开部署，在不同的容器中，需要挂载网络磁盘，这个网络磁盘并不可靠且会受网路带宽限制，性能较差
这个思路还有个问题就是Flumed设置多少条数据进行保存读取位点，设置的大了，容器重启会丢数据，设置小了性能不够，找这个平衡点要耗费大量的时间和资源这个思路因为OpenResty和Flumed在一个容器和不在一个容器都有一些问题，考虑不使用
尝试

尝试golang实现
```
go 1.20.10
gin 1.9.1
sarama 1.41.3
```
我花了几天的时间将其实现，初步性能测试结果如下： 1个CPU核心，1G内存，100并发，每个请求发5个埋点，TPS是731 最终CPU使用率47%，内存使用0.93G 本思路一开始和架构师讨论的时候只是说理论上可行，尝试一下，但谁心里都没底。在收集资料的时候偶然遇到了知乎大佬又拍云的文章【实战分享】使用 Go 重构流式日志网关，有此思路的成功上线的先例，信心大增。
尝试Java实现
```
Spring Boot 3.x（SpringWebFlux版本非SpringMVC版本）
Reactor Kafka 2.x(主要是和kakfa-server对应)
```
1个CPU核心，2G内存，100并发，每个请求5个埋点，TPS是430 最终CPU使用率60%，内存使用1.2G

结论

OpenResty+lua实现的测试结果是 1个CPU核心，1G内存，100并发，每个请求5个埋点，TPS是421 最终CPU使用率60%，内存使用0.6G

根据OpenResty方案来看，Golang和Java实现差距不是特别大，Golang展现的明显的性能优势，但是公司对Golang项目的配套做的并不好，比如实时监控，基础镜像，Golang工程师等。对Java项目比较齐全。目前初步综合考量两个项目均进入UAT环境使用专用压测机进行压测。

压完我再来补充结果。

-----------------------------------20231101分割线 start----------------------------------- 补充测试结果： 1C1G 10w数据 |并发数|TPS-Java实现|TPS-Go实现|CPU使用率-Java实现|CPU使用率-Go实现|内存-Java实现|内存-Go实现| |-|-|-|-|-|-|-| |100-首次|706|1016|峰值100%|峰值100%|<500M|<35M| |100-二次|1021|1015|峰值100%|峰值100%|<500M|<35M| |50|954|933|峰值90%|峰值90%|<500M|<30M| |75|1004|973|峰值84%|峰值90%|<500M|<30M|

总的来看TPS差距不大，CPU使用率上Go的略高，内存上Go实现比Java实现少一个数量级，确实省内存。综合考虑编排发布，公司内部的技术栈深度，监控，日志等周边配套，我们选了Java实现。如果内存资源比较紧张，或者有Golang的技术深度，可以选择Golang的实现。如果内存资源尚可，又是以Java为主的技术栈和周边配套，可以选择Java的实现。 -----------------------------------20231101分割线 end-----------------------------------

以下log一下Sarama向Kafka发消息代码：

var KafkaProduce sarama.AsyncProducer

func InitKafkaConfig() error {
    config := sarama.NewConfig()
    // 配置
    // 等待服务器成功后的响应
    config.Producer.RequiredAcks = sarama.WaitForLocal
    // 随机向partition发送消息
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    // 是否等待成功和失败后的响应，只有上面的RequireAcks设置不是NoReponse这里才有用
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true
    // KafkaClientIpList是[]string类型 值为kafka地址+端口号 一般是3个
    client, err := sarama.NewClient(KafkaClientIpList, config)
    if err != nil {
        return err
    }

    producer, err := sarama.NewAsyncProducerFromClient(client)
    if err != nil {
        return err
    }
    //这个一定要有，不然kafka消息发上一定数量直接就发不动了 
    //原因是你往 KafkaProduce.Input()发消息 会存在本地 不会真正发送到kafka
    //本地开的内存空间用完了 就卡住了
    go func() {
        for {
            select {
            case _ = <-producer.Successes():
            case er := <-producer.Errors():
                if er != nil {
                    AccessLogger.Errorf("Produced message failure: %s", er)
                }
            }
        }
    }()
    KafkaProduce = producer
    return nil
}

func DestroyKafkaProducer() {
    if KafkaProduce != nil {
        KafkaProduce.Close()
    }
}

//消息发送
func SendKafkaAsyncMessage(msg string, topic string) {

    //写入kafka
    KafkaProduce.Input() <- &sarama.ProducerMessage{Topic: topic, Key: nil, Value: sarama.StringEncoder(msg)}
}