kafka精确一次语义EOS的原理深刻剖析-kafka 商业环境实战

时间 2019-11-16

标签 kafka 精确一次语义 eos 原理深刻剖析商业环境实战栏目 Kafka 繁體版

原文原文链接

本套技术专栏是做者（秦凯新）平时工做的总结和升华，经过从真实商业环境抽取案例进行总结和分享，并给出商业应用的调优建议和集群环境容量规划等内容，请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址：1120746959@qq.com，若有任何学术交流，可随时联系。apache

1 Kafka 0.11.0.0版本的逆天之做

0.11.0.0版本以前默认提供at least once语义，想象这样一种场景，分区的Leader副本所在的Broker成功的将消息写入本地磁盘，而后broker将发送响应给producer，此时假设网络出现故障致使该响应没有发送成功。此种状况下，Producer将认为消息发送请求失败，从而开启重试机制。若此时网络恢复正常，那么同一条消息将会被写入两次。
基于上述案例：0.11.0.0版本提供幂等性：每一个分区中精确一次且有序
0.11.0.0版本提供事务：跨分区原子写入机制。

2 故障类型

broker可能故障：Kafka是一个高可用、持久化的系统，每一条写入一个分区的消息都会被持久化而且多副本备份（假设有n个副本）。因此，Kafka能够容忍n-1个broker故障，意味着一个分区只要至少有一个broker可用，分区就可用。Kafka的副本协议保证了只要消息被成功写入了主副本，它就会被复制到其余全部的可用副本（ISR）。
producer到broker的RPC调用可能失败：Kafka的持久性依赖于生产者接收broker的ack响应。没有接收成功ack不表明生产请求自己失败了。broker可能在写入消息后，发送ack给生产者的时候挂了。甚至broker也可能在写入消息前就挂了。因为生产者没有办法知道错误是什么形成的，因此它就只能认为消息没写入成功，而且会重试发送。在一些状况下，这会形成一样的消息在Kafka分区日志中重复，进而形成消费端屡次收到这条消息。
客户端可能会故障：精确一次交付也必须考虑客户端故障。可是咱们如何知道一个客户端已经故障而不是暂时和brokers断开，或者经历一个程序短暂的暂停，区分永久性故障和临时故障是很重要的，为了正确性，broker应该丢弃僵住的生产这发送来的消息，一样，也应该不向已经僵住的消费者发送消息。一旦一个新的客户端实例启动，它应该可以从失败的实例留下的任何状态中恢复，从一个安全点开始处理。这意味着，消费的偏移量必须始终与生产的输出保持同步。

3 Producer幂等性处理机制

若是出现致使生产者重试的错误，一样的消息，仍由一样的生产者发送屡次，将只被写到kafka broker的日志中一次。对于单个分区，幂等生产者不会由于生产者或broker故障而发送多条重复消息。
kafka保存序列号仅仅须要几个额外的字段，所以这种机制的开销很是低。
除了序列号，kafka会为每一个Producer实例分配一个Producer id（PID）,每一条消息都会有序列号，并严格递增顺序。若发送的消息的序列号小于或者等于broker端保存的序列号，那么broker会拒绝这条消息的写入操做。
注意的是：当前的设计只能保证单个producer实例的EOS语义，没法实现多个Producer实例一块提供EOS语义。
想要开启这个特性，得到每一个分区内的精确一次语义，也就是说没有重复，没有丢失，而且有序的语义，只须要设置producer配置中的”enable.idempotence=true”。

4 事务：跨分区原子写入

事务：跨分区原子写入bootstrap

将容许一个生产者发送一批到不一样分区的消息，这些消息要么所有对任何一个消费者可见，要么对任何一个消费者都不可见。这个特性也容许你在一个事务中处理消费数据和提交消费偏移量，从而实现端到端的精确一次语义。安全

主要针对消息通过Partioner分区器到多个分区的状况。网络

producer.initTransactions();
  try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
  } catch(ProducerFencedException e) {
    producer.close();
  } catch(KafkaException e) {
    producer.abortTransaction();
  }
复制代码

5 消费端的事务支持

在消费者方面，有两种选择来读取事务性消息，经过隔离等级“isolation.level”消费者配置表示：session

read_commited：除了读取不属于事务的消息以外，还能够读取事务提交后的消息。
  read_uncommited：按照偏移位置读取全部消息，而不用等事务提交。这个选项相似Kafka消费者的当前语义。
复制代码

为了使用事务，须要配置消费者使用正确的隔离等级。ide
使用新版生产者，而且将生产者的“transactional . id”配置项设置为某个惟一ID。须要此惟一ID来提供跨越应用程序从新启动的事务状态的连续性。fetch

6 消费端精确到一次语义实现

消费端精确到一次语义实现：consumer经过subscribe方法注册到kafka，精确一次的语义要求必须手动管理offset，按照下述步骤进行设置：this

1.设置enable.auto.commit = false;spa
2.处理完消息以后不要手动提交offset，设计
3.经过subscribe方法将consumer注册到某个特定topic，
4.实现ConsumerRebalanceListener接口和consumer.seek(topicPartition,offset)方法（读取特定topic和partition的offset）

5.将offset和消息一块存储，确保原子性，推荐使用事务机制。

public class ExactlyOnceDynamicConsumer {

  private static OffsetManager offsetManager = new OffsetManager("storage2");

  public static void main(String[] str) throws InterruptedException {

      System.out.println("Starting ManualOffsetGuaranteedExactlyOnceReadingDynamicallyBalancedPartitionConsumer ...");

      readMessages();

  }



  private static void readMessages() throws InterruptedException {

      KafkaConsumer<String, String> consumer = createConsumer();

      // Manually controlling offset but register consumer to topics to get dynamically assigned partitions.
      // Inside MyConsumerRebalancerListener use consumer.seek(topicPartition,offset) to control offset

      consumer.subscribe(Arrays.asList("normal-topic"), new MyConsumerRebalancerListener(consumer));

      processRecords(consumer);
  }


  private static KafkaConsumer<String, String> createConsumer() {
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      String consumeGroup = "cg3";

      props.put("group.id", consumeGroup);

      props.put("enable.auto.commit", "false");
      props.put("heartbeat.interval.ms", "2000");
      props.put("session.timeout.ms", "6001");

      * Control maximum data on each poll, make sure this value is bigger than the maximum single record size
      props.put("max.partition.fetch.bytes", "140");

      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      return new KafkaConsumer<String, String>(props);
  }

  private static void processRecords(KafkaConsumer<String, String> consumer) {

      while (true) {

          ConsumerRecords<String, String> records = consumer.poll(100);

          for (ConsumerRecord<String, String> record : records) {

              System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
              offsetManager.saveOffsetInExternalStore(record.topic(), record.partition(), record.offset());

          }
      }
  }
复制代码

}

public class MyConsumerRebalancerListener implements org.apache.kafka.clients.consumer.ConsumerRebalanceListener {

  private OffsetManager offsetManager = new OffsetManager("storage2");
  private Consumer<String, String> consumer;

  public MyConsumerRebalancerListener(Consumer<String, String> consumer) {
      this.consumer = consumer;
  }

  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

      for (TopicPartition partition : partitions) {

          offsetManager.saveOffsetInExternalStore(partition.topic(), partition.partition(), consumer.position(partition));
      }
  }

  public void onPartitionsAssigned(Collection<TopicPartition> partitions) {


      for (TopicPartition partition : partitions) {
          consumer.seek(partition, offsetManager.readOffsetFromExternalStore(partition.topic(), partition.partition()));
      }
  }
复制代码

}

public class OffsetManager {
    private String storagePrefix;
    public OffsetManager(String storagePrefix) {
        this.storagePrefix = storagePrefix;
    }

    void saveOffsetInExternalStore(String topic, int partition, long offset) {

        try {

            FileWriter writer = new FileWriter(storageName(topic, partition), false);

            BufferedWriter bufferedWriter = new BufferedWriter(writer);
            bufferedWriter.write(offset + "");
            bufferedWriter.flush();
            bufferedWriter.close();

        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    long readOffsetFromExternalStore(String topic, int partition) {

        try {

            Stream<String> stream = Files.lines(Paths.get(storageName(topic, partition)));

            return Long.parseLong(stream.collect(Collectors.toList()).get(0)) + 1;

        } catch (Exception e) {
            e.printStackTrace();
        }

        return 0;
    }

    private String storageName(String topic, int partition) {
        return storagePrefix + "-" + topic + "-" + partition;
    }

}
复制代码

6总结

Kafka 0.11.0.0版本的逆天之做，都是在消费者EOS语义较弱，须要进一步加强。

本套技术专栏是做者（秦凯新）平时工做的总结和升华，经过从真实商业环境抽取案例进行总结和分享，并给出商业应用的调优建议和集群环境容量规划等内容，请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址：1120746959@qq.com，若有任何学术交流，可随时联系。

秦凯新于深圳 201812012146