码迷，mamicode.com
首页 > 编程语言 > 详细

kafka集群搭建和使用Java写kafka生产者消费者

时间：2016-10-02 21:35:29      阅读：190      评论：0      收藏：0      [点我收藏+]

标签：

 kafka集群搭建

 

Java代码  技术分享
  1. 1.zookeeper集群  搭建在110， 111,112  
  2.   
  3. 2.kafka使用3个节点110， 111,112  
  4. 修改配置文件config/server.properties  
  5. broker.id=110  
  6. host.name=192.168.1.110  
  7. log.dirs=/usr/local/kafka_2.10-0.8.2.0/logs  
  8. 复制到其他两个节点，然后修改对应节点上的config/server.pro   
  9.   
  10. 3.启动,在三个节点分别执行  
  11. bin/kafka-server-start.sh  config/server.properties >/dev/null 2>&1 &  
  12.   
  13. 4 创建主题  
  14. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test  
  15.   
  16. 5 查看主题详细  
  17. bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test  
  18.  --topic test  
  19. Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:  
  20.         Topic: test     Partition: 0    Leader: 110     Replicas: 110,111,112  Isr: 110,111,112  
  21.         Topic: test     Partition: 1    Leader: 111     Replicas: 111,112,110  Isr: 111,112,110  
  22.         Topic: test     Partition: 2    Leader: 112     Replicas: 112,110,111  Isr: 112,110,111  
  23.   
  24.   
  25. 6 去zk上看kafka集群  
  26. [zk: localhost:2181(CONNECTED) 5] ls /  
  27. [admin, zookeeper, consumers, config, controller, zk-fifo, storm, brokers, controller_epoch]  
  28. [zk: localhost:2181(CONNECTED) 6] ls /brokers   ----> 查看注册在zk内的kafka  
  29. [topics, ids]  
  30. [zk: localhost:2181(CONNECTED) 7] ls /brokers/ids  
  31. [112, 110, 111]  
  32. [zk: localhost:2181(CONNECTED) 8] ls /brokers/ids/112  
  33. []  
  34. [zk: localhost:2181(CONNECTED) 9] ls /brokers/topics   
  35. [test]  
  36. [zk: localhost:2181(CONNECTED) 10] ls /brokers/topics/test   
  37. [partitions]  
  38. [zk: localhost:2181(CONNECTED) 11] ls /brokers/topics/test/partitions  
  39. [2, 1, 0]  
  40. [zk: localhost:2181(CONNECTED) 12]   

 


2  kafka java调用：

 

2.1 java端生产数据， kafka集群消费数据：

 

Java代码  技术分享
  1. 1 创建maven工程，pom.xml中增加如下：  
  2.  <dependency>  
  3.         <groupId>org.apache.kafka</groupId>  
  4.         <artifactId>kafka_2.10</artifactId>  
  5.         <version>0.8.2.0</version>  
  6.     </dependency>  
  7.   
  8.   
  9. 2 java代码：  向主题test内写入数据  
  10.   
  11. import java.util.Properties;  
  12. import java.util.concurrent.TimeUnit;  
  13.   
  14. import kafka.javaapi.producer.Producer;  
  15. import kafka.producer.KeyedMessage;  
  16. import kafka.producer.ProducerConfig;  
  17. import kafka.serializer.StringEncoder;  
  18.   
  19.   
  20.   
  21.   
  22. public class kafkaProducer extends Thread{  
  23.   
  24.     private String topic;  
  25.       
  26.     public kafkaProducer(String topic){  
  27.         super();  
  28.         this.topic = topic;  
  29.     }  
  30.       
  31.       
  32.     @Override  
  33.     public void run() {  
  34.         Producer producer = createProducer();  
  35.         int i=0;  
  36.         while(true){  
  37.             producer.send(new KeyedMessage<Integer, String>(topic, "message: " + i++));  
  38.             try {  
  39.                 TimeUnit.SECONDS.sleep(1);  
  40.             } catch (InterruptedException e) {  
  41.                 e.printStackTrace();  
  42.             }  
  43.         }  
  44.     }  
  45.   
  46.     private Producer createProducer() {  
  47.         Properties properties = new Properties();  
  48.         properties.put("zookeeper.connect", "192.168.1.110:2181,192.168.1.111:2181,192.168.1.112:2181");//声明zk  
  49.         properties.put("serializer.class", StringEncoder.class.getName());  
  50.         properties.put("metadata.broker.list", "192.168.1.110:9092,192.168.1.111:9093,192.168.1.112:9094");// 声明kafka broker  
  51.         return new Producer<Integer, String>(new ProducerConfig(properties));  
  52.      }  
  53.       
  54.       
  55.     public static void main(String[] args) {  
  56.         new kafkaProducer("test").start();// 使用kafka集群中创建好的主题 test   
  57.           
  58.     }  
  59.        
  60. }  
  61.   
  62.   
  63.   
  64.   
  65. 3  kafka集群中消费主题test的数据：  
  66. [root@h2master kafka]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginnin  
  67.   
  68. 4   启动java代码，然后在看集群消费的数据如下：  
  69.   
  70. message: 0  
  71. message: 1  
  72. message: 2  
  73. message: 3  
  74. message: 4  
  75. message: 5  
  76. message: 6  
  77. message: 7  
  78. message: 8  
  79. message: 9  
  80. message: 10  
  81. message: 11  
  82. message: 12  
  83. message: 13  
  84. message: 14  
  85. message: 15  
  86. message: 16  
  87. message: 17  
  88. message: 18  
  89. message: 19  
  90. message: 20  
  91. message: 21  

 

 3 kafka 使用Java写消费者，这样 先运行kafkaProducer ，在运行kafkaConsumer，即可得到生产者的数据：

 

Java代码  技术分享
  1. import java.util.HashMap;  
  2. import java.util.List;  
  3. import java.util.Map;  
  4. import java.util.Properties;  
  5.   
  6. import kafka.consumer.Consumer;  
  7. import kafka.consumer.ConsumerConfig;  
  8. import kafka.consumer.ConsumerIterator;  
  9. import kafka.consumer.KafkaStream;  
  10. import kafka.javaapi.consumer.ConsumerConnector;  
  11.   
  12.   
  13.   
  14.   
  15. /** 
  16.  * 接收数据 
  17.  * 接收到: message: 10 
  18. 接收到: message: 11 
  19. 接收到: message: 12 
  20. 接收到: message: 13 
  21. 接收到: message: 14 
  22.  * @author zm 
  23.  * 
  24.  */  
  25. public class kafkaConsumer extends Thread{  
  26.   
  27.     private String topic;  
  28.       
  29.     public kafkaConsumer(String topic){  
  30.         super();  
  31.         this.topic = topic;  
  32.     }  
  33.       
  34.       
  35.     @Override  
  36.     public void run() {  
  37.         ConsumerConnector consumer = createConsumer();  
  38.         Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
  39.         topicCountMap.put(topic, 1); // 一次从主题中获取一个数据  
  40.          Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);  
  41.          KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据  
  42.          ConsumerIterator<byte[], byte[]> iterator =  stream.iterator();  
  43.          while(iterator.hasNext()){  
  44.              String message = new String(iterator.next().message());  
  45.              System.out.println("接收到: " + message);  
  46.          }  
  47.     }  
  48.   
  49.     private ConsumerConnector createConsumer() {  
  50.         Properties properties = new Properties();  
  51.         properties.put("zookeeper.connect", "192.168.1.110:2181,192.168.1.111:2181,192.168.1.112:2181");//声明zk  
  52.         properties.put("group.id", "group1");// 必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据  
  53.         return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
  54.      }  
  55.       
  56.       
  57.     public static void main(String[] args) {  
  58.         new kafkaConsumer("test").start();// 使用kafka集群中创建好的主题 test   
  59.           
  60.     }  
  61.        
  62. }  

 

 http://chengjianxiaoxue.iteye.com/blog/2190488

kafka集群搭建和使用Java写kafka生产者消费者

标签：

原文地址：http://www.cnblogs.com/lnas01/p/5927958.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！