微信搜索superit|邀请体验:大数据, 数据管理、OLAP分析与可视化平台 | 赞助作者:赞助作者

Kafka 生产者和消费者的笔记

java aide_941 24℃

Kafka 生产者和消费者的笔记
2016年12月23日 11:27:37 梦醉天下 阅读数:11306
版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/dreamsigel/article/details/53836427
Maven依赖:

<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.1.0</version>
        </dependency>

一、生产者

首先先看官方API示例:

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("acks", "all");
 props.put("retries", 0);
 props.put("batch.size", 16384);
 props.put("linger.ms", 1);
 props.put("buffer.memory", 33554432);
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

 Producer<String, String> producer = new KafkaProducer<>(props);
 for(int i = 0; i < 100; i++)
     producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

 producer.close();

设置生产者需要连接的kafka地址,回令类型,重试次数,批量提交大小,提交延迟等待时间(等待时间内可以追加提交),缓存大小,序列化方法。

生产者目前没有碰到太多问题,这也跟kafka一个生产,多个消费的机制有关。需要注意的有两点:

1、acks回令。如果必须等待回令,那么设置acks为all;否则,设置为-1;等待回令会有性能损耗。

2、生产者在发送消息的过程中,会自己默认批量提交。所以,如果单条指令的发送请求,记得发送完后flush才能生效。

二、消费者

依旧先看官方示例:

Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "true");
     props.put("auto.commit.interval.ms", "1000");
     props.put("session.timeout.ms", "30000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("foo", "bar"));
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
             System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
     }

类似生产者的配置,依次设置kafka连接地址,消费者标识(区分消费者offset),是否自动提交offset,提交offset周期,心跳离线超时时间,序列化方法。

消费者问题说明:

1、配置bootstarp.servers后,如非特殊需求,不需要配置zookeeper.server。否则,容易造成zookeeper的数据不一致问题。

2、group.id是标识消费者的ID。每一个group.id消费后,kafka会记录该id消费的offset到zookeeper。所以,此处需要注意,(1)如果多个地方都使用相同的groupid,可能造成个别消费者消费不到的情况(2)如果单个消费者消费能力不足的话,可以启动多个相同groupid的consumer消费,处理相同的逻辑。但是,多线程的时候,需要增加每个groupid下的partition分区数量,便于每个线程稳定读取固定的partition,提高消费能力。

3、如非特殊需求,自动offset提交便可以满足;如果有消息的重复消费需求,可以手动存储每次的offset,自动提交。

4、自动提交周期。消费者每次从队列里拉出来的数量是多条,如果消费能力差,处理速度慢,在提交周期内没有完成处理,那么会导致提交失败,影响消费和系统稳定。所以可以考虑增加周期或者使用内存队列二次存储消息,提升处理速度。另外,可以查看IO是否使用批量提交的方式。如hbase,redis,mysql等。

5、消费者心跳超时时间。如果处理速度超出改时间,kafka认为消费者离线,提交也会失败,也会导致消费问题。所以,需要尽量让心跳超时大于offset自动提交周期。

6、consumer.subcribe(Arrays.asList(“topic1″,”topics”))。消费者可以同时监听多个topic(通道),监听过程属于线程切换。如topic1中拉取10个,再去topic2中拉取10个,一起返回。具体监听topic的数量需要根据消息队列的设置和消息量确定。如果消息很少,一个线程完全可以胜任多个。接受完后增加处理逻辑就可以。如果消息量大,就一个线程一个topic,或者多个线程一个topic。

7、因为kafka会记住消费者的消息处理指针offset,所以,如果一旦消费者离线,那么会造成消息积压,无法保持最新的数据消费。除非每次更换groupid获取最新数据。但是这种更换的方式并不推荐,因为总会产生新的消费者,增加zookeeper的数据。所以,消费者尽量保持在线。如果不使用消息,可以存储、更新或丢弃。

转载请注明:SuperIT » Kafka 生产者和消费者的笔记

喜欢 (0)or分享 (0)