kafka 消息发送和接收

发送代码实例

public class KafkaProducerDemo extends Thread{

    private final KafkaProducer<Integer,String> producer;

    private final String topic;
    private final boolean isAysnc;

    public KafkaProducerDemo(String topic,boolean isAysnc){
        Properties properties=new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "zk01:9092,zk02:9092,zk03:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,"KafkaProducerDemo1");
        properties.put(ProducerConfig.ACKS_CONFIG,"-1");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.IntegerSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        producer=new KafkaProducer<Integer, String>(properties);
        this.topic=topic;
        this.isAysnc=isAysnc;
    }

    @Override
    public void run() {
        int num=0;
        while(num<10){
            String message="message_"+num;
            System.out.println("begin send message:"+message);
            if(isAysnc){//异步发送
                producer.send(new ProducerRecord<Integer, String>(topic, message), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if(recordMetadata!=null){
                            System.out.println("async-offset:"+recordMetadata.offset()+
                                    "->partition"+recordMetadata.partition());
                        }
                    }
                });
            }else{//同步发送  future/callable
                try {
                    RecordMetadata recordMetadata=producer.
                            send(new ProducerRecord<Integer, String>(topic,message)).get();
                    System.out.println("sync-offset:"+recordMetadata.offset()+
                            "->partition"+recordMetadata.partition());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

            }
            num++;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        new KafkaProducerDemo("test",true).start();
    }

消息消费代码示例

public class KafkaConsumerDemo extends Thread{

    private final KafkaConsumer kafkaConsumer;

    public KafkaConsumerDemo(String topic) {
        Properties properties=new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "zk01:9092,zk02:9092,zk03:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"KafkaConsumerDemo1");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.IntegerDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        kafkaConsumer=new KafkaConsumer(properties);
        kafkaConsumer.subscribe(Collections.singletonList(topic));
    }

    @Override
    public void run() {
        while(true){
            ConsumerRecords<Integer,String> consumerRecord=kafkaConsumer.poll(1000);
            for(ConsumerRecord record:consumerRecord){
                System.out.println("message receive:"+record.value());
                kafkaConsumer.commitAsync();
            }
        }
    }

    public static void main(String[] args) {
        new KafkaConsumerDemo("test").start();
    }
}

发送端重要参数说明:
acks :表示 producer 发送消息到 broker 上以后的确认值。有三个可选项
0：表示 producer 不需要等待 broker 的消息确认。这个选项时延最小但同
时风险最大（因为当 server 宕机时，数据将会丢失）。
1：表示 producer 只需要获得 kafka 集群中的 leader 节点确认即可，这个
选择时延较小同时确保了 leader 节点确认接收成功。
all(-1)：需要 ISR 中所有的 Replica 给予接收确认，速度最慢，安全性最高，
但是由于 ISR 可能会缩小到仅包含一个 Replica，所以设置参数为 all 并不能一
定避免数据丢失

batch.size
生产者发送多个消息到 broker 上的同一个分区时，为了减少网络请求带来的
性能开销，通过批量的方式来提交消息，可以通过这个参数来控制批量提交的
字节数大小，默认大小是 16384byte,也就是 16kb，意味着当一批消息大小达
到指定的 batch.size 的时候会统一发送
linger.ms
Producer 默认会把两次发送时间间隔内收集到的所有 Requests 进行一次聚合
然后再发送，以此提高吞吐量，而 linger.ms 就是为每次发送到 broker 的请求
增加一些 delay，以此来聚合更多的 Message 请求。 这个有点像TCP 里面的
Nagle 算法，在 TCP 协议的传输中，为了减少大量小数据包的发送，采用了
Nagle 算法，也就是基于小包的等-停协议

batch.size 和 linger.ms 这两个参数是 kafka 性能优化的关键参数，很多同
学会发现 batch.size 和 linger.ms 这两者的作用是一样的，如果两个都配置
了，那么怎么工作的呢？实际上，当二者都配置的时候，只要满足其中一个要
求，就会发送请求到 broker 上

max.request.size
设置请求的数据的最大字节数，为了防止发生较大的数据包影响到吞吐量，默
认值为 1MB

消费端重要参数说明:

group.id
组 是 kafka 提供的可扩展且具有容错性的消费者机制。一个组内可以有多个消费者或消费者实例consumer instance)，它们共享一个公共的 ID，即 group ID。组内的所有消费者协调一起来消费订阅主题(subscribed topics)的所有分区(partition)。每个分区只能由同一个消费组内的一个 consumer 来消费.如下图所示，分别有三个消费者，属于两个不同的 group，那么对于 firstTopic 这个 topic 来说，这两个组的消费者都能同时消费这个 topic 中的消息，此时，这个 firstTopic 就类似于 ActiveMQ 中的 topic 概念（一条消息可以被多个组同时消费）。如右图所示，如果 3 个消费者都属于同一个group，那么此时 firstTopic 就是一个 Queue 的概念。

enable.auto.commit
消费者消费消息以后自动提交. 当消息提交以后，该消息才不会被再次接收到， 该配置可以配合 auto.commit.interval.ms 控制自动提交的频率.也可以通过 consumer.commitSync()的方式实现手动提交。

auto.offset.reset
对于
这个参数是针对新的 groupid 中的消费者而言的，当有新 groupid 的消费者来消费指定的 topic 时，对于该参数的配置，会有不同的意义
auto.offset.reset=latest 情况下，新的消费者将会从其他消费者最后消费的offset 处开始消费 Topic 下的消息
auto.offset.reset= earliest 情况下，新的消费者会从该 topic 最早的消息开始消费
auto.offset.reset=none 情况下，新的消费者加入以后，由于之前不存在offset，则会直接抛出异常

max.po ll.records
此设置限制每次调用 poll 返回的消息数，这样可以更容易的预测每次 poll 间隔要处理的最大值。通过调整此值，可以减少 poll 间隔