怎么让Kafka达到最佳吞吐量
本篇内容介绍了“怎么让Kafka达到最佳吞吐量”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
我们提供的服务有：成都网站建设、网站制作、微信公众号开发、网站优化、网站认证、化隆ssl等。为数千家企事业单位解决了网站和推广的问题。提供周到的售前咨询和贴心的售后服务，是有科学管理、有技术的化隆网站制作公司
上手使用
func main() {
// 1. 初始化
pusher := kq.NewPusher([]string{
"127.0.0.1:19092",
"127.0.0.1:19092",
"127.0.0.1:19092",
}, "kq")
ticker := time.NewTicker(time.Millisecond)
for round := 0; round < 3; round++ {
select {
case <-ticker.C:
count := rand.Intn(100)
m := message{
Key: strconv.FormatInt(time.Now().UnixNano(), 10),
Value: fmt.Sprintf("%d,%d", round, count),
Payload: fmt.Sprintf("%d,%d", round, count),
}
body, err := json.Marshal(m)
if err != nil {
log.Fatal(err)
}
fmt.Println(string(body))
// 2. 写入
if err := pusher.Push(string(body)); err != nil {
log.Fatal(err)
}
}
}
}将 kafka cluster 配置以及 topic 传入，你就得到一个操作 kafka 的 push operator。
至于写入消息，简单的调用 pusher.Push(msg) 就行。是的，就这么简单！
> 当然，目前只支持单个 msg 写入。可能有人会疑惑，那就继续往下看，为什么只能一条一条写入？
初始化
一起看看 pusher 初始化哪些步骤：
NewPusher(clusterAddrs, topic, opts...) |- kafka.NewWriter(kfConfig) // 与 kf 之前的连接 |- executor = executors.NewChunkExecutor() // 设置内部写入的executor为字节数定量写入
建立与
kafka cluster的连接。此处肯定就要传入kafka config；设置内部暂存区的写入函数以及刷新规则。
使用 chunkExecutor 作用不言而喻：将随机写 -> 批量写，减少 I/O 消耗；同时保证单次写入不能超过默认的 1M 或者自己设定的最大写入字节数。
其实再往 chunkExecutor 内部看，其实每次触发插入有两个指标：
maxChunkSize：单次最大写入字节数flushInterval：刷新暂存消息插入的间隔时间
在触发写入，只要满足任意一个指标都会执行写入。同时在 executors 都有设置插入间隔时间，以防暂存区写入阻塞而暂存区内消息一直不被刷新清空。
> 更多关于 executors 可以参看以下：https://zeromicro.github.io/go-zero/executors.html
生产者插入
根据上述初始化对 executors 介绍，插入过程中也少不了它的配合：
func (p *Pusher) Push(v string) error {
// 1. 将 msg -> kafka 内部的 Message
msg := kafka.Message{
Key: []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
Value: []byte(v),
}
// 使用 executor.Add() 插入内部的 container
// 当 executor 初始化失败或者是内部发生错误，也会将 Message 直接插入 kafka
if p.executor != nil {
return p.executor.Add(msg, len(v))
} else {
return p.produer.WriteMessages(context.Background(), msg)
}
}过程其实很简单。那 executors.Add(msg, len(msg)) 是怎么把 msg 插入到 kafka 呢？
插入的逻辑其实在初始化中就声明了：
pusher.executor = executors.NewChunkExecutor(func(tasks []interface{}) {
chunk := make([]kafka.Message, len(tasks))
// 1
for i := range tasks {
chunk[i] = tasks[i].(kafka.Message)
}
// 2
if err := pusher.produer.WriteMessages(context.Background(), chunk...); err != nil {
logx.Error(err)
}
}, newOptions(opts)...)触发插入时，将暂存区中存储的
[]msg依次拿出，作为最终插入消息集合；将上一步的消息集合，作为一个批次插入
kafka的topic中
这样 pusher -> chunkExecutor -> kafka 一个链路就出现了。
“怎么让Kafka达到最佳吞吐量”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！
文章名称：怎么让Kafka达到最佳吞吐量
URL地址：http://cdiso.cn/article/jcgshe.html
