Kafka Scala Producer API
2016-06-28 22:51
155 查看
Kafka Sca
4000
la Producer API
官方example:https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example
构造Producer
构造一个kafka.javaapi.producer.Producer<K, V>对象,K:分区key类型,V:消息类型
Properties props = new Properties(); props.put("metadata.broker.list", "broker1:9092,broker2:9092");//该配置项必须有,producer通过连接这些broker找到partition的leader,只要是集群中的broker就行,不一定要leader的 props.put("serializer.class", "kafka.serializer.StringEncoder");//序列化,这个类的输入决定了KeyedMessage<K,V>中V的类型,默认为kafka.serializer.DefaultEncoder,即直接传输byte[] props.put("partitioner.class", "example.producer.myPartitioner");//分区规则,也有默认的,可以自定义 props.put("request.required.acks", "1");//发送消息成功后需要等待kafka的回应 ProducerConfig config = new ProducerConfig(props); Producer<String, String> producer = new Producer<String, String>(config);
构造消息
构造消息kafka.producer.KeyedMessage<K,V>,K V与上述相同
KeyedMessage<String, String> data = new KeyedMessage<K, V>(String topic, K key, V message); //若无kay参数,将随机投递至一个分区 KeyedMessage<String, String> data = new KeyedMessage<K, V>(String topic, V message);
发送消息
producer.send(data);
关闭producer
producer.close();
相关文章推荐
- 主机无法访问linux虚拟机中tomcat的服务-解决办法
- Android 使用Socket实现服务器与手机客户端的长连接七:定义消息体,区分是响应还是回复信息
- 算术平均、几何平均、平方平均、调和平均
- Linux掉电处理
- 如何调试PHP的Core之获取基本信息 --------风雪之隅 PHP7核心开发者
- 使用IntelliJ IDEA 14和Maven创建java web项目
- Android 使用Socket实现服务器与手机客户端的长连接七:定义消息体,区分是响应还是回复信息
- python 学习简记 《编程导论》 CH6&CH7
- Struts2中Action需要用Servlet API的方法
- 使用Eclipse连接hbase1.1 hadoop2.6
- 用php实现边执行边输出的效果
- android 关机对话框添加重启功能
- css 样式使用包括样式文件导入
- Eclipse相关配置
- CSS3阴影全接触
- uva 272
- WWDC Session视频没有字幕解决办法
- JavaSE 基础 第10节 初识字符串
- Amazon的分布式存储引擎Dynamo
- java filechannel