
kafka的应用场景

- 实时日志聚合，支持高容量事件流
- 处理大量的数据积压，支持来自离线系统的周期性数据加载
- 处理低延迟分发，分区，分布式，以及实时处理



### 如何处理大量的数据积压？

熟知的mysql是通过B+树来持久化到磁盘的，查找数据需要从磁盘读取并构建B+树内存数据结构。而Kafka则是：使用linux文件系统和pagecache，放弃内存数据结构

- 即使进程维护了内存数据结构，该数据也可能会被复制到操作系统的 pagecache 中，事实上所有内容都被存储了两份。

- 通过自动访问所有空闲内存将可用缓存的容量至少翻倍，并且通过存储紧凑的字节结构而不是独立的对象，有望将缓存容量再翻一番。 这样使得32GB的机器缓存容量可以达到28-30GB,并且不会有额外的 GC 负担
- 保持 内存cache 和 文件系统之间一致性逻辑，不需要做过多的转化。

- 简单的读写文件，读写操作不会相互影响。数据的大小不在会影响性能。可以无限量的使用物理硬盘，价格便宜，容量充足。可以充分解决大量的数据积压难题



### 如何在1秒内接收并持久化这1亿条数据呢？

存储问题解决了，面临的问题：大量的I/O和字节拷贝

I/O操作主要是 客户端与服务端的消息传递，服务端的数据持久化。

将多个消息打包成一组消息块，一次请求发送一组消息，服务端接收，consumer每次获取多组消息块，大量的I/O操作将不在是主要的性能问题，其实主要矛盾是在服务带宽问题上。producer/broker/consumer 三者使用二进制数据传输，producer 和 consumer 之间使用相同的压缩算法也可以极大的节省带宽问题

sendfile 将数据从 pagecache 直接发送到网络，这样避免重新复制数据。



每秒1亿条数据，保存7天，kafka是如何处理的？

概述

- kafka 库
- producer config 配置
- sync vs async
- 压缩方式

背景：博客文章想统计访问数据，每日pv/uv/ip数据

分析：

- 网页浏览数据，主要以写为主
- 时效性、事务持久要求不高
- 数据量大

kafka特别适合这种场景，高吞吐量，较低的端到端延迟，强大的持久性。

# go kafka 库

- [github.com/Shopify/sarama](https://pkg.go.dev/github.com/Shopify/sarama)  使用人数较多，我公司也是用的这个库
- [github.com/confluentinc/confluent-kafka-go](https://github.com/confluentinc/confluent-kafka-go) 我没有做过实验，有兴趣的朋友自己尝试一下

```
go get github.com/Shopify/sarama
```

# sync vs async

同步异步的区别, 如果因为实时性要求没那么高的话，推荐使用async，总结一下原因：kafka面临的2大难题之一，**持久化要面临磁盘大量的I/O操作**，如此频繁的写，即便是顺序写、有pagecache可以缓冲、有mmap减少一次内核态的复制，磁盘也是难以承受。

kafka的处理也是很简单，加缓存，本来1毫秒100次写，broker缓存10次一写（数据块），这样可以减轻10倍的I/O压力。

所以这里也是同样的思想，kafka加了一层缓存，我们producer也可以在加一层缓存，又可以减轻n倍的I/O压力。所以在实时性要求不高的前提下，推荐使用async

async有4个配套参数

```go
conf := sarama.NewConfig()
conf.Producer.Flush.Messages = 10                      // 缓存条数
conf.Producer.Flush.Frequency = 500 * time.Millisecond // 缓存时间

Flush struct {
  // 达到多少字节时，触发一次broker请求
  Bytes int
  // 达到多少条消息时，触发一次broker请求
  Messages int 
  // producer缓存消息的时间, 默认缓存500毫秒 => queue.buffering.max.ms
  Frequency time.Duration 
  // producer缓存队列里最大缓存的消息数量 => queue.buffering.max.messages
  MaxMessages int
}
```

完整的prodcuer用法

```go
type Produce struct {
	AsyncProducer sarama.AsyncProducer
}

func NewProducer(ctx context.Context) *Produce {
	conf := sarama.NewConfig()
	conf.Producer.RequiredAcks = sarama.WaitForAll
	conf.Producer.Return.Errors = true                     // 接收producer的error，下面细说用法
	conf.Producer.Compression = sarama.CompressionLZ4      // 压缩方式，如果kafka版本大于1.2，推荐使用zstd压缩
	conf.Producer.Flush.Messages = 10                      // 缓存条数
	conf.Producer.Flush.Frequency = 500 * time.Millisecond // 缓存时间

	producer, err := sarama.NewAsyncProducer(config.Kafka.Broker(), conf)
	if err != nil {
		panic(errors.Wrapf(err, "failed to open kafka producer: "))
	}
	pd := &Produce{AsyncProducer: producer}
	// 接受结束信号
	go func() {
		for {
			select {
			case err := <-producer.Errors():
				logger.Errorf("kafka producer send error %s", err.Err.Error())
			case <-ctx.Done(): // 接收主进程结束信号，持久化缓存数据，防止丢失，关闭producer连接
				_ = producer.Close()
				logger.Infof("quit: kafka producer")
				return
			}
		}
	}()
	return pd
}

func (p *Produce) Send(topic string, value []byte) {
	p.AsyncProducer.Input() <- &sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.ByteEncoder(value),
	}
}
```

- `conf.Producer.Return`: 这里是go语言的专属用法，有2个channel，**重点：**如果 `conf.Producer.Return.Errors = true` 设置为了true，就必须有相应读取channel，否则会阻塞producer写消息，success channel/error channel默认capacity只有256字节 <a name="producer_return"></a>
    - `success`: 写消息成功channel 
    -  `errors`:  错误channel
- `conf.Producer.RequiredAcks` 对应java producer的 `request.required.acks`
    - `sarama.NoResponse` - `0`:   producer不等待来自broker同步完成的确认继续发送下一条（批）消息。延迟最低/消息持久化保证性最低（当服务器发生故障时某些数据会丢失，如leader已死，但producer并不知情，发出去的信息broker就收不到）。
    - `sarama.WaitForLocal` - `1`: producer在leader已成功收到的数据并得到确认后发送下一条message，没有确认其他副本是否收到。持久化保证较好（被写入死亡leader但尚未复制将失去了唯一的消息）
    - `sarama.WaitForAll` - `2`（默认）:  producer在leader以及follower副本确认接收到数据后才算一次发送完成。 

- `conf.Producer.Compression` : kafka 消息压缩了，综合考量来说，如果kafka版本大于2.1可以使用zstd，小于2.1版本可以使用lz4, 无论是cpu占用还是压缩比例的角度上看

| 算法           | cpu  | 压缩比（带宽同理） | producer吞吐量(压缩快慢) | consumer吞吐量(解压缩快慢) |
| -------------- | ---- | ------------------ | ------------------------ | -------------------------- |
| Gzip           | 1    | 3                  | 4                        | 3                          |
| Snappy         | 4    | 4                  | 2                        | 2                          |
| LZ4            | 3    | 2                  | 1                        | 1                          |
| ZSTD (2.1版本) | 2    | 1                  | 3                        | 4                          |

# issue

1. go kafka producer在写消息时，写了1，2条就阻塞了？ 

答：[conf.Producer.Return配置姿势不对？](#producer_return) 

`conf.Producer.Return`: 这里是go语言的专属用法，有2个channel，**重点：**如果 `conf.Producer.Return.Errors = true` 设置为了true，就必须有相应读取channel，否则会阻塞producer写消息，success channel/error channel默认capacity只有256字节 <a name="producer_return"></a>

- `success`: 写消息成功channel 
-  `errors`:  错误channel



go kafka producer在写消息时，写了1，2条就阻塞了？

go kafka producer 实践与落地

prometheus/jmx_exporter 收集jmx指标，grafana 监控 kafka

分享 kafka prometheus 监控方案: jmx 监控指标 grafana dashboard


概述：

- kafka消费者、消费者组和消费者位移？
- kafka幂等性以及避免重复消费？
- go kafka consumer 封装

背景：和producer同样的用途，统计博客网页浏览数据

# kafka消费者、消费者组和消费者位移？

kafka consumer 从 broker 中 pull 数据，producer push 数据到 broker 中。

优点：consumer可以承受能力的范围内控制消费速度，而且可以批量获取消息块

缺点：如果broker中没有数据，consumer会有轮询broker接口，这里kafka提供了参数来阻塞fetch请求直到有数据可以消费

**消费者组的特性：**

1. 一个消费者组可以有一个或者多个消费者
2. topic的一个partition只能有一个消费者组中的一个消费者消费，要想实现分布式消费，必须要有多个patition

**消费者位移：**

kafka面临的问题，如何让broker 和 consumer 就被消费的数据保持一致性

- broker发送消息给consumer之后立即标记为consumed。如果 consumer 崩溃或者请求超时或者其他原因，就会导致消息丢失
- 增加确认机制，消息发送标记消息为sent，consumer消费成功后标记消息为consumed。
    - 如果consumer消费成功了，但是在发送确认之前崩溃了，broker需要再次发送消息，就会导致消息重复消费。
    - 性能问题：broker需要为每个消息保存多个状态，状态的变动需要加锁
    - 如果消息一直得不到确认该如何处理

kafka的是如何实现消息确认机制的：

每一个 partition 在任意给定的时间内只能被每个订阅了这个 topic 的 consumer 组中的一个 consumer 消费，只需要保存这个consumer下一条要消费的消息的位移即可

- 同一个消费者组的消费者 不会并发消费 同一个partition，不需要加锁
- 不需要保存多个状态，消费者确认要消费到了哪一条，只想下一条即可。
- 如果consumer消费逻辑出现bug，可以在 bug 修复后通过回退到之前的 offset 来再次消费这些数据 （重要）, go 可以指定出初始位移`conf.Consumer.Offsets.Initial`来再次消费



# kafka幂等性以及避免重复消费

正常情况下producer push一条消息到broker，然后consumer从broker pull消息然后消费，这只是理想状态的情况下。我们需要考虑的是发生异常的情况kafka如何做到保证消息是不丢失，不重复消费的。

- producer 发布消息的持久性保证
- consumer 消费消息的保证，只被消费一次

**如果一个 producer 在试图发送消息的时候发生了网络故障， 则不确定网络错误发生在消息提交之前还是之后，broker到底有没有收到这条消息并持久化**？

答：如果我们想要确保消息可以重传，但绝不丢失。kafka 0.11.0.0 之前的版本就是重发机制，之后的版本增加了幂等性，会为每条消息分配序列号来避免在broker log中产生重复消息。同时kafka提供了事务机制，多条消息写入log要么都成功要么都失败。

kafka所有的副本都有相同的 log 和相同的 offset。consumer 负责控制它在 log 中的位置，理想状态下consumer永远不会崩溃，没问题。**如果consumer崩溃了，这个topic partition必须要有另一个进程来接管，此时该从哪个offset开始接管？**

1. consumer先读取消息，然后将它的位置保存在log中，在处理相应的消息，如果异常发生在位置保存后，即使有其他进程接管，这条消息可能也会丢失了
2. consumer先读取消息，处理消息，最后保存它的位置，如果异常是在消息处理完成后，还没有保存消息时。消息可能会被重复处理。

答：2种解决方案，步骤就是先读取，在消费，最后更新位移，这样可以保证消息不丢失，可能会造成消息重复消费

- 简单的方案：幂等性保证，使用offset作为唯一键，数据和offset一起写库，如果offset不存在就先建，否则更新。
- 复杂的方案：2步提交，消费完成写库，状态为待提交，更新kafka log offset，更新库的状态为已提交，此方案会复杂化问题，推荐幂等性保证，简单有效。

重点：consumer消费数据时，要做一次幂等性消费，避免重复消费数据



# kafka 消费者 & 分区

Kafka 消费者从属于消费者群组。一个群组里的消费者订阅的是同一个主题，每个消费者 接收主题一部分分区的消息。

- 假设主题 T1 有 4 个分区，我们创建了消费者 C1，它是群组 G1 里唯一的消费者，我们用 它订阅主题 T1。消费者 C1 将收到主题 T1 全部 4 个分区的消息
- 如果在群组 G1 里新增一个消费者 C2，那么每个消费者将分别从两个分区接收消息。假设消费者 C1 接收分区 0 和分区 2 的消息，那么消费者 C2 节会接收分区 1 和分区 3 的消息
- 如果群组 G1 有 4 个消费者，那么每个消费者可以分配到一个分区
- 如果群组添加更多的消费者，超过主题的分区数量，那么有一部分消费者就会被 闲置，不会接收到任何消息

**再均衡**：分区的所有权从一个消费者转移到另一个消费者

消费者通过向consumer-group 协调器的broker发送心跳来维持他们和群组的从属关系以及它们对分区的所有权关系，只要消费者以正常的时间间隔发送心跳，就被认为是活跃的，说明还在正常读取分区的消息，

如果消费者发生崩溃，consumer-group协调器broker，在确认了该consumer死亡后会出发再均衡。

`max.poll.interval.ms` 控制发送心跳的间隔时间

`session.timeout.ms` 控制失联多久判定为死亡

# go kafka consumer 

封装一下go consumer支持

- 优雅关闭consumer
- 记录consumer异常，同producer一样Consumer.Return channel设置为true以后必须有相应的读取，否则会阻塞
- kafka崩溃恢复，可重新加入consumer group，恢复消费

```go
type ConsumerGroup interface {
	Consume(topic []string, handler sarama.ConsumerGroupHandler)
}

type consumerGroup struct {
	ctx     context.Context
	client  sarama.ConsumerGroup
	groupId string
}

func NewConsumerGroup(ctx context.Context, groupId string) ConsumerGroup {
	conf := sarama.NewConfig()
	conf.Version = sarama.V0_11_0_2
	conf.Consumer.Return.Errors = true
	client, err := sarama.NewConsumerGroup(config.Kafka.Broker(), groupId, conf)
	if err != nil {
		panic(errors.Wrap(err, "kafka new consumer client err"))
	}
	// Track errors
	go func() {
		for err := range client.Errors() {
			logger.Wrapf(err, "kafka %s consume err", groupId)
		}
	}()
	go func() {
		for {
			select {
			// 主进程退出，通知consumer关闭
			case <-ctx.Done():
				_ = client.Close()
				//logger.Infof("quit: kafka consumer %s", groupId)
				return
			}
		}
	}()
	return &consumerGroup{
		ctx:     ctx,
		client:  client,
		groupId: groupId,
	}
}

func (c *consumerGroup) Consume(topic []string, handler sarama.ConsumerGroupHandler) {
	go func() {
		defer c.client.Close()
		for {
			err := c.client.Consume(c.ctx, topic, handler)
			if err != nil {
				switch err {
				case sarama.ErrClosedClient, sarama.ErrClosedConsumerGroup:
					// 退出
					logger.Infof("quit: kafka consumer %s", c.groupId)
					return
				case sarama.ErrOutOfBrokers:
					logger.Errorf("kafka 崩溃了~")
				default:
					logger.Errorf("kafka exception: %s", err.Error())
				}
				time.Sleep(1 * time.Second)
			}
		}
	}()
}
```
---

go kafka consumer group 使用姿势


[toc]

# 安装

**zookeeper** 下载地址: [http://zookeeper.apache.org/releases.html#download](http://zookeeper.apache.org/releases.html#download)

解压：`tar -zxvf kafka_2.12-2.6.0.tgz`

运行：`./bin/zkServer.sh --config conf start`

> --config 指的是配置目录



**kafka** 下载地址: [https://kafka.apache.org/downloads](https://kafka.apache.org/downloads)

解压：`tar -zxvf kafka_2.12-2.6.0.tgz`

运行：`./bin/kafka-server-start.sh config/server.properties`

# 配置

**kafka配置：**

`kakfa-configs.sh` 可以对kafka配置查看或修改

别名： 每次敲这么多命令很烦

`alias kconfigs="~/Documents/kafka/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092"`

查看:  --describe

`kconfigs --describe --entity-type [topics/clients/users/brokers/broker-loggers] --all`

- brokers
  - log.dirs：指定文件存储目录, 多个逗号分隔，建议目录指定不同的磁盘
    - 提升读写性能，多个磁盘同时读写的性能更高
    - 故障转移：Failover 1.1 版本添加
  - listeners：通过什么协议访问kafka服务，内网服务指定
  - advertised.listeners：外网访问kafka服务指定
  - log.retension.hours=168：记录保留时长 默认7天（hours/minutes/ms）
  - log.retension.bytes=1073741824：可用磁盘大小 默认 128G， -1不限制
  - zookeeper.connect=localhost:2181：zookeeper连接地址
  - auto.create.topics.enable: false topic不存在时是否自动创建（建议显示配置false）
  - unclean.leader.election.enblae: false 允许落后的副本参与leader选举（建议显示配置false）
  - auto.leader.rebalance.enable: false 定期leader选举，正常服务的leader到期也会被替换掉（建议显示配置false）
- topics
  - retension.hours 保留时长
  - retension.bytes 为该Topic预留多大的磁盘空间，-1无限使用磁盘空间
- users
- clients
- broker-loggers

修改：--alter

`kconfigs --alter --entity-type topics --entity-name test --add-config max.message.bytes=104858800`

- --alter：代表修改
- --entity-type: 代表上面这5中实例
- --entity-name: 指定那个broker（broker.id）, topic (topic.name)
- --add-config: --help 可以查看支持实时修改的配置

**JVM配置**：

- 堆大小(heap size)配置：建议6G  环境变量
  - `export KAFKA_HEAP_OPTS=--Xms6g  --Xmx6g`
- 垃圾回收：G1收集器 
  - `export  KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true`

**系统配置**:

- 文件描述符限制：`ulimit -n`
  - `/etc/security/limits.conf` 
    - `soft nofile 102400`
    - `hard nofile 104800`
- 文件系统类型： xfs 性能较好
- swap内存：建议配置成1，方便报警
- 脏页刷新时间：默认5秒

- 磁盘大小
  - 新增消息数
  - 消息留存时间
  - 平均消息大小
  - 备份数
  - 是否启用压缩

# 本地集群

本地运行多个实例，需要指定不同的端口，和broker.id, 创建多个server.propertie，指定端口

```
config/server-1.properties:
  broker.id=1
  listeners=PLAINTEXT://:9093
  log.dir="～/Documents/kafka/data1"
```

```
config/server-2.properties:
  broker.id=2
  listeners=PLAINTEXT://:9094
  log.dir="~/Documents/kafka/data2"
```

启动：

```shell
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
./bin/kafka-server-start.sh -daemon config/server.properties
./bin/kafka-server-start.sh -daemon config/server1.properties 
./bin/kafka-server-start.sh -daemon config/server2.properties 
```

创建：3副本/1分区 topic

> alias ktopics="~/Documents/kafka/bin/kafka-topics --bootstrap-server localhost:9092,localhost:9093,localhost:9094"

```shell
ktopics --create --replication-factor 3 --partitions 1 --topic test
```

查看：topic状态

```shell
ktopics --descibe --topic test

Topic: test	PartitionCount: 1	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 3	Replicas: 2,3,1	Isr: 3,1,2
```

# kafka 分区策略 & 压缩算法

分区策略是决定生产者将消息发送到哪个分区的算法

常见的分区策略:

1. 轮询策略，能保证消息尽可能分配到不同的分区上 
2. 随机策略，随机返回小于partition的数，不能保证均匀性，适用及其性能不一的情形
3. hash策略，具有相同key的消息会保证在同一个分区中，这样可以保证消息的前后顺序

go语言包：[github.com/Shopify/sarama](https://pkg.go.dev/github.com/Shopify/sarama)  提供了这3种分区策略的实现

**轮询:** `func NewRoundRobinPartitioner(topic string) Partitioner`

**随机:** `func NewRandomPartitioner(topic string) Partitioner`

**hash:** `func NewHashPartitioner(topic string) Partitioner`

**指定分区** `func NewManualPartitioner(topic string) Partitioner`



压缩：

保证Producer 和 Consumer 压缩算法一致

```go
const (
	//CompressionNone no compression
	CompressionNone CompressionCodec = iota
	//CompressionGZIP compression using GZIP
	CompressionGZIP
	//CompressionSnappy compression using snappy
	CompressionSnappy
	//CompressionLZ4 compression using LZ4
	CompressionLZ4
	//CompressionZSTD compression using ZSTD
	CompressionZSTD
)
```

吞吐量： LZ4 > Snappy > zstd > GZIP

压缩比：zstd > LZ4 > GZIP > Snappy

CPU:  各算法差不多压缩时 Snappy 算法使用的 CPU 较多, 解压缩时 GZIP 算法可能使用更多的 CPU



# FAQs

**如何保证消息不丢失?**

producer: 

1. 消息发送异常处理, [sarama](https://github.com/Shopify/sarama)提供了2种发送消息的方式，sync/async，同步如果发送失败会返回error，异步提供了 `Errors() <-chan *ProducerError` 接口来处理错误
2. 消息发送异常重试, `config.Producer.Retry`, 默认是 3次
3. `config.Producer.RequiredAcks=WaitForAll` 所有副本都接受到消息，该消息才算做提交
   1. `NoResponse RequiredAcks = 0`
   2. `WaitForLocal RequiredAcks = 1`
   3. `WaitForAll RequiredAcks = -1`

broker:

1. `unclean.leader.election.enable=false` 控制落后台太多的broker不参加leader选举
2. `replication.factor >= 3` 副本数量不要少于3份
3. `min.insync.replicas >= 1` 消息最少写入多少个副本才算做已提交，正式环境中建议>=3
4. `replication.factor > min.insync.replicas`  如果2着相等，只要有1个副本挂掉，整个分区就无法工作了

consumer:

1. 先消费，在更新位移, `cfg.Consumer.Offsets.AutoCommit.Enable = false` 默认是自动提交的


kafka生产环境配置


- kafka 多线程处理单个消费者消息
- kafka 单线程处理单个消费者消息
- kafka 增加分区数量

# kafka 多线程处理单个消费者消息

写php的时候经常拿redis来当做消息队列，然后启动多个进程处理队列中的消息，失败的消息记录写入失败队列中。所以在处理kafka consumer group时首先会想到这种方式来处理消息。只是如果kafka也按照这种方式来的话，保证消息不丢失的语义将会是一件麻烦事。

位移: kafka partition 只会维护一个数字，即处理的最后一条消息。

如果我们启用多线程来处理同一个消费者的消息的话，考虑一下这种情况:

4个线程处理，分别拿到了 位移 5,6,7,8 的消息

- 拿到8的线程成功处理了消息。但是5，6，7的线程处理失败了，此时我们还需要一个队列来处理失败的消息。就像上面处理redis队列一样
- 还是拿到8的线程成功处理了消息，提交了位移，5的线程最后处理完消息，最后提交commit，此时6，7，8的消息会被重复消费

所以我们多线程处理单个消费者消息是需要解决这处理失败，消息重复消费这2个问题，不符合kafka消息交付语义。实现链路长，代码复杂

**那么为什么还是需要多线程消费？**

答：考虑需要使用消息队列场景，比如：视频、图片处理

- 处理周期长，超过web请求设定的超时时间
- 异步处理，业务需求更关注处理结果，对时间有一定包容度
- 服务降级，比如文章阅读统计但是还参杂算法推荐，运营活动，但是不要影响阅读量统计的逻辑。毕竟吞吐量要求比较高

java 实现代码可以看  [胡夕老师的多线程实践](https://www.cnblogs.com/huxi2b/p/13668061.html) 

go实现代码

```go

type Consume struct {
	ctx     context.Context
	encoder encode.Encoder
	// sem 信号量，限制并发数
	sem chan struct{}
}

// n: goroutine 并发数量
func NewPdfConvertConsume(ctx context.Context, encoder encode.Encoder, n int) *Consume {
	return &Consume{
		ctx:     ctx,
		encoder: encoder,
		sem:     make(chan struct{}, n),
	}
}

func (*Consume) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (*Consume) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

func (p *Consume) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for m := range claim.Messages() {
		p.sem <- struct{}{}
		go func(msg sarama.ConsumerMessage) {
			defer func() {
				<-p.sem
			}()
			data := &action.Context{}
			logger.Debugf("consumer msg: %s", string(msg.Value))
			if err := p.encoder.Decode(data, msg.Value); err != nil {
				logger.Wrapf(err, "msg decoder failed")
			} else {
        // todo 业务逻辑
			}
		}(*m)
		session.MarkMessage(m, "")
	}
	return nil
}

```

# kafka 单线程处理单个消费者的消息

- 可以保证kafka的消息交付语义的，实现方式简单易维护
- 需要更多的TCP连接

- 线程自己处理消息可能会造成超时导致rebalance

如果消息处理时间较长，可以调整参数 或者 细化处理步骤保证消息处理速度 来避免频繁的rebalance

**参数调优**:

`heartbeat.interval.ms` : 心跳间隔，用来保持consumer的会话，并且在有consumer加入或者离开group时帮助进行rebalance，这个值必须要小于 `session.timeout.ms`, 在超过`session.timeout.ms` 时间内如果没有收到hearbeat消息，就会将该consumer移出 consumer group, 一般设置 `session.timeout.ms `的 1/3

`session.timeout.ms` ： session会话过期时间

`max.poll.interval.ms` : 最大poll数据间隔，默认值是3s, 如果超过这个时间还是没有发起poll请求，即使heartbeat依旧在发，还是会把consumer 移出 consumer group



# kafka 增加分区数

```
./kafka-topics.sh --zookeeper localhost:2181/kafka --alter --topic demo --partitions 6
```



1.kafka 多线程处理单个消费者消息 2. kafka 单线程处理单个消费者消息 3. kafka 增加分区数量