且构网

分享程序员开发的那些事...
且构网 - 分享程序员编程开发的那些事

kafka权威指南 第三章第3节 创建Producer

更新时间:2022-10-02 23:18:17

问题导读:
1 如何创建Kafka的生产者?
2 发送数据都有哪几种方式?

kafka权威指南 第三章第3节 创建Producer


向Kafka中写入数据的第一步就是创建生产者对象,创建生产者对象需要三个主要的配置:

bootstrap.servers
是Kafka的Broker的连接地址,格式为 host:port。这里不要填写所有的broker配置,生产者会查询这些broker的信息,然后添加额外的broker。但是推荐至少填写两个,以免有一个broker离线后,无法连接到集群。
key.serializer
Kafka broker 接收的是消息的键值对的字节数组,但是生产者的接口也支持参数,来发送java对象。这样代码的可读性更高,但是Java对象也需要序列化成字节数组才行。key.serializer需要设置成类的名字,并且这个类需要实现org.apache.kafka.common.serialization.Serializer接口,生产者将会使用这个类对对象的key进行序列化。Kafka的客户端提供了ByteArraySerializer,StringSerializer以及IntegerSerializer,你可以直接使用这些通用的类型,如果有需要的话,也可以自定义。注意,key.serializer只有在你发送一个值的时候有用。
value.serializer
与key.serializer类似,也是配置一个类的名字,然后将会把消息的value部分进行序列化。你的key和value可以设置成不同的序列化格式,如整型的key和字符串的value。

下面是创建Producer客户端的例子,一般情况下你都可以使用这些默认的配置:
private Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "broker1:9092,broker2:9092");
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<String, String>(kafkaProps);

1 第一行,创建属性配置对象;
2 然后为key和value设置它的序列化格式
3 最后根据配置创建Producer

有了这样一个简单的接口,producer的配置就算完成了。官网文档里面介绍了更多配置的参数,我们将在后面进行更详细的介绍。

一旦启动了这个producer,它就会开始发送消息。有下面三种机制:
Fire and forget 
这种模式会发送给broker消息,但是不关心消息是否正常到达。大多数情况下,它会正常到达,有时候也会在失败的时候尝试多发送几次。但是需要注意的是,这种情况下消息会有丢失
同步发送
这种发送的机制是使用send()方法,它会返回一个对象,通过get()方法可以同步等待,来判断是成功还是失败
异步发送
同样也是调用send,但是会在Kafka broker接收到消息的时候返回。

在上面所有的情况下,都需要考虑发送数据失败的情况以及如何处理它,并且注意单独的producer对象可以使用多线程来发送数据或者也可以使用多个producer。因此,如果你需要更高的吞吐量,你可以在相同的prodcuer上创建更多的线程,也可以开启更多的producer。

在后续的例子中,将会使用我们提到的方法来创建producer,并且处理不同情况可能发生的错误。

来自: http://www.aboutyun.com/forum.php?mod=viewthread&tid=22001 

本文转自   ChinaUnicom110   51CTO博客,原文链接:http://blog.51cto.com/xingyue2011/1939981