kafka (java API demo)
2017-04-06 16:30
344 查看
版本:kafka0.10.2
Producer API
生产者API允许应用程序发送数据流到kafka集群中的topic。
介绍怎么使用producer的例子在这个文档里javadocs
如果你使用maven构建工程的话,要使用producer,你需要在pom文件添加下面的依赖:
Consumer API
消费者API允许应用程序从Kakfa集群中的topic中读取数据流。
介绍怎么使用producer的例子在这个文档里javadocs
consumer需要添加的依赖与producer相同:
Streams API
streams API可以改造数据流并输入输出到topics
在javadocs中有详细的例子
Streams API可供使用的额外的文档在这里
要使用kafka streams 需要在pom文件中添加下面的依赖:
Producer API
生产者API允许应用程序发送数据流到kafka集群中的topic。
介绍怎么使用producer的例子在这个文档里javadocs
如果你使用maven构建工程的话,要使用producer,你需要在pom文件添加下面的依赖:
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.10.2.0</version> </dependency>
Consumer API
消费者API允许应用程序从Kakfa集群中的topic中读取数据流。
介绍怎么使用producer的例子在这个文档里javadocs
consumer需要添加的依赖与producer相同:
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.10.2.0</version> </dependency>
Streams API
streams API可以改造数据流并输入输出到topics
在javadocs中有详细的例子
Streams API可供使用的额外的文档在这里
要使用kafka streams 需要在pom文件中添加下面的依赖:
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-streams</artifactId> <version>0.10.2.0</version> </dependency>
一、生产者
package kafka; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; public class UserKafkaProducer extends Thread{ private final KafkaProducer<Integer,String> producer; private final String topic; private final Properties props=new Properties(); public UserKafkaProducer(String topic){ // props.put("metadata.broker.list", ""); props.put("bootstrap.servers", "172.16.31.105: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=new KafkaProducer<Integer,String>(props); this.topic=topic; } @Override public void run() { int messageNo=1; while (messageNo<5){ String messageStr=new String("Message_"+messageNo); System.out.println("Send:"+messageStr); producer.send(new ProducerRecord<Integer, String>(topic, messageStr)); messageNo++; try { sleep(3000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } }
二、消费者
package kafka; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; public class UserKafkaConsumer extends Thread{ private final ConsumerConnector consumer; private final String topic; 4000 public UserKafkaConsumer(String topic){ consumer=kafka.consumer.Consumer.createJavaConsumerConnector( createConsumerConfig()); this.topic=topic; } private ConsumerConfig createConsumerConfig() { Properties props=new Properties(); props.put("zookeeper.connect","172.16.31.101:2181,172.16.31.102:2181,172.16.31.103:2181"); props.put("group.id","test-consumer-group"); props.put("zookeeper.session.timeout.ms","40000"); props.put("zookeeper.sync.time.ms","200"); props.put("auto.commit.interval.ms","1000"); return new ConsumerConfig(props); } @Override public void run() { Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); topicCountMap.put(topic, new Integer(1)); Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0); ConsumerIterator<byte[], byte[]> it = stream.iterator(); while (it.hasNext()) { System.out.println("receive:" + new String(it.next().message())); try { sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
三、启动运行
package kafka; public class UserKafkaTest { public static void main(String[] args) { UserKafkaProducer producerThread=new UserKafkaProducer(KafkaProperties.topic); producerThread.start(); UserKafkaConsumer consumerThread=new UserKafkaConsumer(KafkaProperties.topic); consumerThread.start(); } }
四、可以将配置提取出来
package kafka; public interface KafkaProperties { final static String zkConnect = "172.16.31.101:2181,172.16.31.102:2181,172.16.31.103:2181"; final static String groupId="test-consumer-group"; final static String topic ="topic1"; final static String mytopic="mytopic"; final static String kafkaServerURL = "172.16.31.105"; final static int kafkaServerPort = 9093; final static int kafkaProducerBufferSize = 64 * 1024; final static int connectionTimeOut = 20000; final static int reconnectInterval = 10000; final static String topic2 = "topic2"; final static String topic3 = "topic3"; final static String clientId = "SimpleConsumerDemoClient"; }
相关文章推荐
- kafka2.9.2的分布式集群安装和demo(java api)测试
- kafka2.9.2的分布式集群安装和demo(java api)测试
- kafka2.9.2的分布式集群安装和demo(java api)测试
- kafka2.9.2的伪分布式集群安装和demo(java api)测试
- kafka2.9.2的分布式集群安装和demo(java api)测试
- kafka 简单安装以及java小demo
- KafkaSparkDemoMain
- kafka-3python生产者和消费者实用demo
- Java API获取topic所占磁盘空间(Kafka 1.0.0)
- 简单封装hbase的Java api工具类 demo,以及hbase的部分ColumFamily压缩验证
- Kafka+Storm+HBase项目Demo(1)--CDH搭建hadoop集群
- Kafka+Storm+HBase项目Demo(7)--Trident使用
- hbase java api 使用demo
- kafka+storm集成并运行demo-单机
- Kafka+Storm+HBase项目Demo(3)--Storm安装配置
- kafka java 生产消费程序demo示例
- Java API 生产和消费Kafka消息
- Kafka 生产者和消费者 demo (java&scala)
- Kafka java 实现消息队列demo
- kafka java 实现消息队列demo