kafka consumer 的配置

Posted 2022-03-31 mrrightzhao

fetch.min.bytes.   #获取最小字节数据
Consumer 向broker中要数据时是按大小来返回的，如果数据没有达到指定的M数，consumer会处于等待状态，直到broker 从producer 哪里获取到指定大小的数据为止。获取取的最小数据大小是指的每个partition上的数据。
fetch.max.wait.ms
当consumer 一直在等待broker的数据达到最小字节数时，如果一直没有达到，consumer是不会一直处于block状态的，会根据fetch.max.wait.ms 设置的时间取结束block 状态。假如说设置fetch.min.bytes 的大小是1m,但是一直没有达到1m，但是到了fetch.max.wait.ms 设置的最大等待时间时会结束掉当前的block状态，有多少数据，获取多少数据。默认值是500ms.
通过这2个参数来控制客户端和服务端的通讯次数，而且还能进行批量作业。
max.partition.fetch.bytes
控制每1个partition最大的字节数。默认是1mb
假如1个topic中有20个partitions,有5个consumer读取partition上的数据，每个consumer只能读取到4M的concumer records.
session.timeout.ms
consumer和broker之间的会话超时时间
假如consumer长时间没有发起poll的信息，没有拿过元数据，并且在session.timeout.ms 的规定时间内，就可以认为客户端超时，如果broker认为consumer超时了会引起consumer 的rebalance,导致其它consmer此时不会收到消息，会将超时的consumer的partition分配给其它的consumer
heartbeat.interval.ms
控制consumer 多长时间发送一次心跳。
因此这2个参数经常放在一起使用，heartbeat.interval.ms 肯定比session.timeout.ms的时间短。一般情况设置心跳的时间是超时时间的1/3 是比较好的。
auto.offset.reset  #earliest
自动从最开始的地方读取数据
当一个topic中有数据时，会通过group.id将读到的数据指定到分组中，当有新的分组创建时，可以通过设置该属性从最开始的地方读取数据。如果没有创建新的分组是读取不到数据的
enable.auto.commit
自动提交    #默认是true 
当consumer读了多少数据的offset，其实是由client端维护的，client端会将offset提交给broker，然后是由专门的线程自动为我们进行处理的。
假如在读取数据时，程序意外死亡(断电、宕机)，它没有来得及提交给broker,会导致下次再次读数据时会读到重复的数据，因为kafka没有记录你最新读取数据的位置，显然这要比你没有读取到数据好的多。所以我们需要的是不管我发送多少次，你只能收到一次，这就牵扯到了幂等性的问题，在后面会介绍。
auto.commit.interval.ms 1000 #设置自动提交时间
partition.assignment.strategy
Partition的分配策略
当consumer读topic中的一个或若干个tipic中partition数据数据时，consumer到底会到哪个partition。在broker中有2个分配策略。1.range 区间   2.roundRobin 
1.range 区间
   当partition出现奇数据的时候会出现分布不均的情况。
   假如有2个topic t1 t2,分别有3个partition,当有2 个consumer时，c1 会被分配到t1下的p1 p2 和c2 下的t1和t2, c2 会被分配到t1 下的p3 和t2 下的p3.这种分配策略是不会将所的的patition分配给一个consumer .这是kafka默认的配置 
2.roundRobin
           它会提供比range更好的分配策略。
         它会根据message 取模的方式将partition 平均的分配给每个topic下面的partition
   对于这2种分配策略，你可以自己通过实现接口的方式实现自定义的分配策略。但是通常情况我们是不会重写consumer的分配策略
cliend.id  #客户端id
    任何合法的string都可以,可以通过这个id在broker查看是哪个客户端发送的信息
max.poll.records  #最大获取数据大小
    假如我们在feach.min.bytes 中设置最小获取10M的数据，而在这里设置最大获取5M，那么consumer会将
剩下的5M缓存起来，再次调用时会将缓存中的数据返回
receive.buffer.bytes
    
send.buffer.bytes