为什么要封装
kafka官方自带的客户端,需要对参数进行设置,如下代码,很多参数的key都是字符串,这样对于编程人员来说非常不友好。参数很多的时候,有两种处理方式:(1)传一个config类进去解析;(2)使用建造者模式,笔者在此就使用建造者模式,对官方客户端进行简单封装,使之易用。
客户端封装demo
官方的例子如下:
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:4242");
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();
封装之后:
CloudKafkaConsumer<String, String> kafkaConsumer =
KafkaConsumerBuilder.builder()
.withBootstrapServers("172.31.1.161:9092")
.withGroupId("connector")
.withAutoCommit("true")
.withCommitIntervalMs("1000")
.withSessionimeoutMs("30000")
.build();
kafkaConsumer.subscribe(Arrays.asList("test"));
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(5000);
for (ConsumerRecord<String, String> record : records)
System.out.printf(record.value());
}
这样其实还不是很好,对于value的类型没有限定,所以我们可以对value的类型限定。如果是基础类型就限定为:boolean、int、String。如果不是基础类型就限定为Enum,这样客户端使用起来会优雅很多。