Kafka学习分享(1)

1-基本概念
2-生产者的使用
3-了解flume中的kafka sink的使用

基本概念

Kafka的几个使用的场景和特性
1-消息系统
Kafka和传统的消息系统都具备系统解耦,冗余存储,流量削峰,缓冲,异步通信等功能。
同时,Kafka还提供了大多数消息系统难以实现的消息顺序性保障以及回溯消费的功能。
2-存储系统
Kafka把消息存储到磁盘
消息持久化功能和多副本机制,我们可以把Kafka作为长期的数据存储系统来使用。
3-Kafka和流式处理紧密结合,比如Spark Streaming+kafka ,flink+kafka。
4-副本机制
Kafka为每个分区引入了副本机制,通过增加副本的数量可以提升容灾能力。
副本之间是一主多从的关系,其中leader副本负责处理读写请求,follower副本只负责
与leader副本的消息同步。副本处于不同的broker中,当leader副本出现故障,从follower
副本中重新选举新的leader副本对外提供服务。

生产者的使用

开发Kafka生产者需要使用的依赖,这里使用的版本是:<kafka.version>0.9.0.1</kafka.version>
      <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.10</artifactId>
        <version>${kafka.version}</version>
      </dependency>

开发的步骤如下:
  配置生产者客户端的参数
  构建需要发送的消息
  发送消息
  关闭生产者

 private static Producer<String, String> createProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.88.129:9092");
        props.put("acks", "all");
        //这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
        props.put("retries", 0);
        // Controls how much bytes sender would wait to batch up before publishing to Kafka.
        //控制发送者在发布到kafka之前等待批处理的字节数。 满足batch.size和ling.ms之一，producer便开始发送消息
        //默认16384   16kb
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        /*自定义分区配置
        props.put("partitioner.class","com.fayayo.study.kafka.partition.CoustomPartitioner");*/
        return new KafkaProducer(props);
    }

private static void sendMessages(Producer<String, String> producer) {
        System.out.println("start send message......");
        String topic = "test-topic";
        int partition = 1;//指定分区发送
        long record = 1;
        for (int i = 1; i <= 200; i++) {
            //指定分区发送
            /*producer.send(
                    new ProducerRecord<String, String>(topic, partition,
                            Long.toString(record),"producer_msg"+Long.toString(record++)),new SendCallBack());*/
            //不指定分区,会均匀发送
            producer.send(
                    new ProducerRecord<String, String>(topic,"producer_msg"+Long.toString(record++)));
        }
        System.out.println("start send message......end");
    }

上面实现了简单的案例。接下来我们来更进一步的了解一下KafkaProducer.
KafkaProducer是线程安全的,可以在多线程当中共享单个KafkaProducer实例.
针对消息的发送提供了三种模式:
1-发后即忘
2-同步
3-异步

要实现同步的方式可以使用producer.send(record).get();
实际上send()方法本身就是异步的,send()方法返回的Future对象可以使调用方稍后获取发送结果。
上面的代码直接在send()方法后调用了get()方法来阻塞等待Kafka的响应。
异步发送一般是在send()方法中指定一个Callback的回调函数。我们会在最后通过分析Kafka Sink的使用时看到这种使用方式。

KafkaProducer发送消息到broker的过程中会需要经过拦截器,序列化,分区器等作用。拦截器不是必须的,暂时不去考虑。
序列化是指:
生产者需要用序列化器把对象转换成字节数组才能通过网络发送给Kafka。
而在对侧,消费者需要用反序列化器把从Kafka中收到的字节数组转换成对应的对象。

分区器
kafka发送的消息其实是一个ProducerRecord。如果消息ProducerRecord中指定了partition字段,那么就不需要分区器的作用,因为partition代表的就是所发往的分区。如果ProducerRecord中没有指定了partition字段,那么就需要依赖分区器,根据key这个字段
来计算partition的值。
Kafka提供了默认的分区器,当然我们也可以自己实现一个分区器。

常见的参数:
1-acks
acks=1:生产者发送消息后只需要分区的leader副本成功写入,那么它就会收到来自服务端的成功响应。
如果消息写入leader副本并返回成功响应给生产者,且在被其他follower副本拉取之前leader副本崩溃,
那么此时消息还是会丢失。

acks=0:生产者发送消息后不等待任何服务端的相应。

acks=-1:有比较强的可靠性,需要等待所有副本都成功写入消息之后才能收到来自服务端的响应。

2-max.request.size:
这个参数用来限制生产者客户端能发送的消息的最大值。默认1M。
不建议盲目增大。因为此参数涉及到一些联动参数。比如broker端的message.max.bytes.

了解flume中的kafka sink的使用

在KafkaSink中,我们可以看到完整的KafkaProducer的使用

1-configure方法
这里进行配置生产者客户端的参数
private void setProducerProps(Context context, String bootStrapServers) {
    kafkaProps.put(ProducerConfig.ACKS_CONFIG, DEFAULT_ACKS);//对应partition的leader写到本地后即返回成功
    //Defaults overridden based on config
    kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, DEFAULT_KEY_SERIALIZER);
    kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DEFAULT_VALUE_SERIAIZER);
    kafkaProps.putAll(context.getSubProperties(KAFKA_PRODUCER_PREFIX));
    kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
  }

2- start方法
producer = new KafkaProducer<String,byte[]>(kafkaProps);

3-process方法
这里实现真正的消息发送,可以只关注kafka的部分
for (; processedEvents < batchSize; processedEvents += 1) {
        //从channel获取event
        event = channel.take();
        
        ...

        //构造需要发送的event
        byte[] eventBody = event.getBody();
        Map<String, String> headers = event.getHeaders();

        eventTopic = headers.get(TOPIC_HEADER);
        if (eventTopic == null) {
          eventTopic = topic;
        }
        eventKey = headers.get(KEY_HEADER);
        
        ...

        // create a message and add to buffer
        long startTime = System.currentTimeMillis();

        //获取是否配置了分区
        Integer partitionId = null;
        try {
          ProducerRecord<String, byte[]> record;
          if (staticPartitionId != null) {
            partitionId = staticPartitionId;
          }
          //Allow a specified header to override a static ID
          if (partitionHeader != null) {
            String headerVal = event.getHeaders().get(partitionHeader);
            if (headerVal != null) {
              partitionId = Integer.parseInt(headerVal);
            }
          }
          //构造需要发送的消息,指定分区和不指定分区 
          if (partitionId != null) {
            record = new ProducerRecord<String, byte[]>(eventTopic, partitionId, eventKey,
                serializeEvent(event, useAvroEventFormat));
          } else {
            record = new ProducerRecord<String, byte[]>(eventTopic, eventKey,
                serializeEvent(event, useAvroEventFormat));
          }
          //异步发送消息,指定了callback
          kafkaFutures.add(producer.send(record, new SinkCallback(startTime)));
        } catch (NumberFormatException ex) {
          throw new EventDeliveryException("Non integer partition id specified", ex);
        } catch (Exception ex) {
          // N.B. The producer.send() method throws all sorts of RuntimeExceptions
          // Catching Exception here to wrap them neatly in an EventDeliveryException
          // which is what our consumers will expect
          throw new EventDeliveryException("Could not send event", ex);
        }
      }

      //Prevent(防止) linger.ms from holding the batch,当达到配置的批次数量后，直接刷新，该方法会将数据全部生产到Kafka
      producer.flush();


4-callback
当消息异步发送成功或者失败的时候会调用当前的这个callback.
class SinkCallback implements Callback {
  private static final Logger logger = LoggerFactory.getLogger(SinkCallback.class);
  private long startTime;

  public SinkCallback(long startTime) {
    this.startTime = startTime;
  }

  public void onCompletion(RecordMetadata metadata, Exception exception) {
    if (exception != null) {
      logger.debug("Error sending message to Kafka {} ", exception.getMessage());
    }

    if (logger.isDebugEnabled()) {
      long eventElapsedTime = System.currentTimeMillis() - startTime;
      logger.debug("Acked message partition:{} ofset:{}",  metadata.partition(), metadata.offset());
      logger.debug("Elapsed time for send: {}", eventElapsedTime);
    }
  }
}

后续

kafkaConsumer的使用
kafkaConsumer在flume中的使用