kafka生产者

生产者

生产者如何发送消息到broker

1. 生产者调用send(ProducerRecord)方法。ProducerRecord 是 Kafka 中的一个核心类，它代表了一组 Kafka 需要发送的 key/value 键值对，它由记录要发送到的主题名称（Topic Name），可选的分区号（Partition Number）以及可选的键值对构成。
2. 然后经过拦截器 -》 序列化器(kafak自己的序列化器。java的序列化器太重了。有一部分是保证数据安全的数据。)
3. 经过分区器。在分区器决定向那个分区里面发送。1. 在发送消息的时候可以指定分区2. 未指定分区 1. 存在key的情况下，将key的hash值与topic的分区数进行取余得到了要发送的分区值。2. 也不存在key的情况下，kafka采用Sticky Partition(粘性分区),会随机选择一个分区，并尽可能的一直使用该分区。待该分区的batch已满或者已完成，kafka再随机一个分区进行使用(和上一次的分区不同)3. 如果key不存在，则根据metadata.max.age.ms这个参数的时间范围随机选择一个。对于这个时间段内，都会发送到唯一的分区里面。这个值默认情况下十分钟更新一次。
4. 消息追加到收集器RecoderAccumulator。1. 消息收集器RecoderAccumulator为每个分区创建一个双端队列Deque<ProducerBatch>2. ProducerBatch 可以理解为是 ProducerRecord 的集合，批量发送有利于提升吞吐量，降低网络影响。3. 由于生产者客户端使用 java.io.ByteBuffer 在发送消息之前进行消息保存，并维护了一个 BufferPool 实现 ByteBuffer 的复用，该缓存池只针对特定大小（ batch.size指定）的 ByteBuffer进行管理，对于消息过大的缓存，不能做到重复利用。每次追加一条ProducerRecord消息，会寻找/新建对应的双端队列，从其尾部获取一个ProducerBatch，判断当前消息的大小是否可以写入该批次中。若可以写入则写入；若不可以写入，则新建一个ProducerBatch，判断该消息大小是否超过客户端参数配置 batch.size 的值，不超过，则以 batch.size建立新的ProducerBatch，这样方便进行缓存重复利用；若超过，则以计算的消息大小建立对应的 ProducerBatch ，缺点就是该内存不能被复用了。4. 每个队列中有多个批次，单个批次大小=Max(batch.size,单条消息的大小)。5. 缓冲区的数据如果准备好，会唤醒sender后台线程去发送数据
5. sender线程将分区队列缓冲的数据读出来之后发送到kafka集群中(Broker)1. 该线程是一个后台线程，是KafkaProducer初始化的时候创建成功的。2. 只有满足batch.size或者linger.ms会唤醒sender线程发送数据3. 如何发送数据。 1. 从消息收集器获取缓存的消息，处理成 <Node, List<ProducerBatch>的形式。node表示broker的节点。相当于每个broker的消息汇总到一块。2. 后面会进一步封装成<Node, Request>形式，此时才可以向服务端发送数据。3. 在发送之前，Sender线程将消息以 Map<NodeId, Deque> 的形式保存到InFlightRequests 中进行缓存，可以通过其获取 leastLoadedNode ,即当前Node中负载压力最小的一个，以实现消息的尽快发出。4. 如果第一个请求发送到broker1中，且没有得到应答。是允许发送第二个请求的。最多允许五个请求未应答。
6. Selector是打通链路的作用，一边相当于输入流一边相当于输出流。1. kafka接收到数据之后，存在一个副本同步的机制。存在一个应答acks 1. 0:生产者在成功写入消息之前是不会等待任何的来自服务器的响应。也就是说如果当中出现了错误，导致broker没有收到消息，那么生产者是无从得知的2. 1:只要集群的首领节点收到消息，生产者就会收到来自服务器成功的响应。若果消息不能够被首领节点接收（比如说首领节点崩溃，而新的首领尚未选出来），这时候生产者会收到一个错误响应3. all / -1:只有在集群所有的跟随副本都接收到消息后，生产者才会受到一个来自服务器的成功响应。这种模式是最安全的。具体等待的副本数量配置参数为min.insvnc.replicas2. 当生产者收到消息为成功的时候。会先清理到请求，然后清理掉分区中每个队列中发送完成的数据。3. 失败的话，会重试，默认是int的最大值(2147483647)。可配置 retries

生产者如何提高吞吐量

• batch.size和linger.ms配合使用，如batch.size=32K、linger.ms=5-100ms
• 在发送数据的时候采用压缩的方式。compression.type:压缩shappy
• 修改缓冲区大小。RecordAccumulator。buffer.memory

消息的可靠性

• 0: 生产者发送过来的数据，不需要等待数据落盘即应答。丢失数据风险大
• 1: 生产者发送过来的数据，Leader收到数据后应答。丢失数据。- 应答完成还没有同步副本，Leader挂了。从新选举的Leader无法获取原先的消息。
• -1: 生产者发送过来的数据，Leader和ISR队列里面的所有副本节点同步数据之后应答。一般都是Follower主动去Leader拉取数据- 当存在一个Follower节点挂了，那就迟迟无法收到ACK的应答信息。- Leader维护了一个动态的in-sync replica set(ISR)。Leader保持和同步的Follower+Leader集合(leader:= 0,follower: 0,1, 2).- 如果Follower长时间未向Leader发送通信求情和同步数据，则该Follower将被提出ISR。该时间阈值由replica.lag.time.max.ms参数设置，默认30s。假设2超时，(leader:= 0,follower: 0,1).- 如果分区副本设置为1个，或者ISR里应答的最小副本数量(min.instnc.replicas 默认为1)设置为1，和ack=1的效果是一样的，仍然存在丢失数据的风险。(leader:= 0,follower: 0).

如何做到完全的数据可靠性

ACK级别设置为-1+分区副本大于等于2+ISR里应答的最小副本数量大于等于2。

数据去重

ack=-1级别的时候，leader接收成功，与副本也同步完成。在准备应答的时候Leader挂了。此时ACK没有成功，会从副本里面选举出一个Leader.。此时消息会重新发送。因为新选举的Leader上一次已经同步到数据，这次又接收一次。导致数据重复。

• 至少一次=ACK级别设置为-1+分区副本大于等于2+ISR里应答的最小副本数量大于等于2。保证数据不丢失，但是无法保证数据重复
• 最多一次=ACK=0 保证数据不重复但是无法保证数据不丢失
• 精确一次=幂等性+至少一次 - 幂等性是kafka0.11版本以后引入的新特性 - 重复数据判断的依据：具有<PID ,Partition,SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是kafka每次重启都会分配一个新的；Partition表示分区号；Sequence Number是单调递增的。幂等性只能保证是在单分区单会话(每次启动都是一次新的会话)不重复。- 触发幂等性的时候，直接在内存中就丢弃消息，不会进行落盘- 通过配置 enable.idempotence 默认为true,false关闭

生产者事务

开启事务，必须开启幂等性。事务依赖幂等性

场景

1. 最简单的需求是producer发的多条消息组成一个事务这些消息需要对consumer同时可见或者同时不可见 。
2. producer可能会给多个topic，多个partition发消息，这些消息也需要能放在一个事务里面，这就形成了一个典型的分布式事务。
3. 不同于rocketmq，rocketmq是保证本地事务和MQ消息发送的事务一致性，kafka是保证了一次发送多条消息的事务一致性。
4. kafka的应用场景经常是应用先消费一个topic，然后做处理再发到另一个topic，这个consume-transform-produce过程需要放到一个事务里面，比如在消息处理或者发送的过程中如果失败了，消费位点也不能提交。
5. producer或者producer所在的应用可能会挂掉，新的producer启动以后需要知道怎么处理之前未完成的事务 。
6. 流式处理的拓扑可能会比较深，如果下游只有等上游消息事务提交以后才能读到，可能会导致rt非常长吞吐量也随之下降很多，所以需要实现read committed和read uncommitted两种事务隔离级别。

事务的配置

• Broker configs - transactional.id.timeout.ms :在ms中，事务协调器在生产者TransactionalId提前过期之前等待的最长时间，并且没有从该生产者TransactionalId接收到任何事务状态更新。默认是604800000(7天)。这允许每周一次的生产者作业维护它们的id- max.transaction.timeout.ms:事务允许的最大超时。如果客户端请求的事务时间超过此时间，broke将在InitPidRequest中返回InvalidTransactionTimeout错误。这可以防止客户机超时过大，从而导致用户无法从事务中包含的主题读取内容。默认值为900000(15分钟)。这是消息事务需要发送的时间的保守上限。- transaction.state.log.replication.factor :事务状态topic的副本数量。默认值:3- transaction.state.log.num.partitions :事务状态主题的分区数。默认值:50- transaction.state.log.min.isr:事务状态主题的每个分区ISR最小数量。默认值:2- transaction.state.log.segment.bytes:事务状态主题的segment大小。默认值:104857600字节
• Producer configs - enable.idempotence :开启幂等- transaction.timeout.ms:事务超时时间事务协调器在主动中止正在进行的事务之前等待生产者更新事务状态的最长时间。这个配置值将与InitPidRequest一起发送到事务协调器。如果该值大于max.transaction.timeout。在broke中设置ms时，请求将失败，并出现InvalidTransactionTimeout错误。默认是60000。这使得交易不会阻塞下游消费超过一分钟，这在实时应用程序中通常是- transactional.id:用于事务性交付的TransactionalId。这支持跨多个生产者会话的可靠性语义，因为它允许客户端确保使用相同TransactionalId的事务在启动任何新事务之前已经完成。如果没有提供TransactionalId，则生产者仅限于幂等交付。
• Consumer configs - isolation.level - read_uncommitted:以偏移顺序使用已提交和未提交的消息。- read_committed:仅以偏移量顺序使用非事务性消息或已提交事务性消息。为了维护偏移排序，这个设置意味着我们必须在使用者中缓冲消息，直到看到给定事务中的所有消息。

publicstaticvoidmain(String[] args)throwsInterruptedException,ExecutionException,TimeoutException{Map<String,Object> configs =newHashMap<>();// 设置连接Kafka的初始连接用到的服务器地址,如果是集群，则可以通过此初始连接发现集群中的其他broker
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");// 设置key的序列化器
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,IntegerSerializer.class);// 设置value的序列化器
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);// 启用幂等性 
        configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);//指定事务配置    必须要配置  事务ID
        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,UUID.fastUUID());KafkaProducer<Integer,String> producer =newKafkaProducer<Integer,String>(configs);//初始化事务
        producer.initTransactions();//启动事务
        producer.beginTransaction();try{Future<RecordMetadata> send = producer.send(newProducerRecord<Integer,String>("topic_3","TransactionsProducer123"));//            int i = 1 / 0;//提交事务
            producer.commitTransaction();}catch(Exception e){
            e.printStackTrace();//终止事务
            producer.abortTransaction();}finally{//关闭生产者
            producer.close();}}//在事务内提交已经消费的偏移量（主要用于消费者）voidsendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata> offsets,String consumerGroupId)throwsProducerFencedException;

数据乱序

每个broker最多缓存5个请求，如果当前两个请求没有应答成功。但是第三个请求应答成功，这时候数据就会出现乱序。第三个请求的数据排到前两个请求之前。但是不管如何，也只能保证单分区消息有序，无法保证全局有序。

• kafka在1.x版本之前保证单分区有序。max.in.flight.requests.per.connection=1 (不需要考虑是否开启幂等性) 每个broker的请求缓存队列设置为1
• kafka在1.x版本之后保证数据单分区有序有两种方式 - 未开启幂等性 max.in.flight.requests.per.connection=1- 开启幂等性 max.in.flight.requests.per.connection 小于5 - 在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，所以都可以保证最近5个request的数据都是有序的。假设发送第三条数据的时候，失败了。那么在服务端只会先落盘消息1和消息2.消息4和消息5不会落盘。会等到消息3拿到之后才会落盘。(幂等性存在一个递增ID)

kafka生产者参数详解

参数解释acks:至少要多少个分区副本接收到了消息返回确认消息 一般是 0:只要消息发送出去了就确认(不管是否失败) 1:只要 有一个broker接收到了消息 就返回 all： 所有集群副本都接收到了消息确认 当然 2 3 4 5 这种数字都可以， 就是具体多少台机器接收到了消息返回， 但是一般这种情况很少用到buffer.memory：生产者缓存在本地的消息大小 ： 如果生产者在生产消息的速度过快 快过了往 broker发送消息的速度 那么就会出现buffer.memory不足的问题 默认值为32M 注意 单位是byte 大概3355000左右max.block.ms:生产者获取kafka元数据(集群数据，服务器数据等) 等待时间 ： 当因网络原因导致客户端与服务器通讯时等待的时间超过此值时 会抛出一个TimeOutExctption 默认值为 60000msretries：设置生产者生产消息失败后重试的次数 默认值 3次retry.backoff.ms:设置生产者每次重试的间隔 默认 100msbatch.size:生产者批次发送消息的大小 默认16k 注意单位还是bytelinger.ms:生产者生产消息后等待多少毫秒发送到broker 与batch.size 谁先到达就根据谁 默认值为0compression.type：kafka在压缩数据时使用的压缩算法 可选参数有:none、gzip、snappy none即不压缩 gzip,和snappy压缩算法之间取舍的话 gzip压缩率比较高 系统cpu占用比较大 但是带来的好处是 网络带宽占用少， snappy压缩比没有gzip高 cpu占用率不是很高 性能也还行， 如果网络带宽比较紧张的话 可以选择gzip 一般推荐snappyclient.id:一个标识， 可以用来标识消息来自哪， 不影响kafka消息生产max.in.flight.requests.per.connection：指定kafka一次发送请求在得到服务器回应之前,可发送的消息数量request.timeout.ms客户端等待请求响应的最大时长。如果服务端响应超时，则会重发请求，除非达到重试次数。该设置应该比 replica.lag.time.max.ms (a brokerconfiguration)要大，以免在服务器延迟时间内重发消息。int类型值，默认：30000，可选值：[0,…]interceptor.classes