漫游kafka实战篇之搭建Kafka开发环境

2022-10-05 12:08:15

上篇文章中我们搭建了kafka的服务器，并可以使用Kafka的命令行工具创建topic，发送和接收消息。下面我们来搭建kafka的开发环境。
 
添加依赖
 
搭建开发环境需要引入kafka的jar包，一种方式是将Kafka安装包中lib下的jar包加入到项目的classpath中，这种比较简单了。不过我们使用另一种更加流行的方式：使用maven管理jar包依赖。
创建好maven项目后，在pom.xml中添加以下依赖：
  1. <dependency>
  2. <groupId> org.apache.kafka</groupId >
  3. <artifactId> kafka_2.10</artifactId >
  4. <version> 0.8.0</ version>
  5. </dependency>
添加依赖后你会发现有两个jar包的依赖找不到。没关系我都帮你想好了，点击这里下载这两个jar包，解压后你有两种选择，第一种是使用mvn的install命令将jar包安装到本地仓库，另一种是直接将解压后的文件夹拷贝到mvn本地仓库的com文件夹下，比如我的本地仓库是d:\mvn,完成后我的目录结构是这样的：
 
漫游kafka实战篇之搭建Kafka开发环境
 
配置程序
 
首先是一个充当配置文件作用的接口,配置了Kafka的各种连接参数：
  1. package com.sohu.kafkademon;
  2. public interface KafkaProperties
  3. {
  4. final static String zkConnect = "10.22.10.139:2181";
  5. final static String groupId = "group1";
  6. final static String topic = "topic1";
  7. final static String kafkaServerURL = "10.22.10.139";
  8. final static int kafkaServerPort = 9092;
  9. final static int kafkaProducerBufferSize = 64 * 1024;
  10. final static int connectionTimeOut = 20000;
  11. final static int reconnectInterval = 10000;
  12. final static String topic2 = "topic2";
  13. final static String topic3 = "topic3";
  14. final static String clientId = "SimpleConsumerDemoClient";
  15. }
producer
 
  1. package com.sohu.kafkademon;
  2. import java.util.Properties;
  3. import kafka.producer.KeyedMessage;
  4. import kafka.producer.ProducerConfig;
  5. /**
  6. * @author leicui bourne_cui@163.com
  7. */
  8. public class KafkaProducer extends Thread
  9. {
  10. private final kafka.javaapi.producer.Producer<Integer, String> producer;
  11. private final String topic;
  12. private final Properties props = new Properties();
  13. public KafkaProducer(String topic)
  14. {
  15. props.put("serializer.class", "kafka.serializer.StringEncoder");
  16. props.put("metadata.broker.list", "10.22.10.139:9092");
  17. producer = new kafka.javaapi.producer.Producer<Integer, String>(new ProducerConfig(props));
  18. this.topic = topic;
  19. }
  20. @Override
  21. public void run() {
  22. int messageNo = 1;
  23. while (true)
  24. {
  25. String messageStr = new String("Message_" + messageNo);
  26. System.out.println("Send:" + messageStr);
  27. producer.send(new KeyedMessage<Integer, String>(topic, messageStr));
  28. messageNo++;
  29. try {
  30. sleep(3000);
  31. } catch (InterruptedException e) {
  32. // TODO Auto-generated catch block
  33. e.printStackTrace();
  34. }
  35. }
  36. }
  37. }
 
consumer
 
  1. package com.sohu.kafkademon;
  2. import java.util.HashMap;
  3. import java.util.List;
  4. import java.util.Map;
  5. import java.util.Properties;
  6. import kafka.consumer.ConsumerConfig;
  7. import kafka.consumer.ConsumerIterator;
  8. import kafka.consumer.KafkaStream;
  9. import kafka.javaapi.consumer.ConsumerConnector;
  10. /**
  11. * @author leicui bourne_cui@163.com
  12. */
  13. public class KafkaConsumer extends Thread
  14. {
  15. private final ConsumerConnector consumer;
  16. private final String topic;
  17. public KafkaConsumer(String topic)
  18. {
  19. consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
  20. createConsumerConfig());
  21. this.topic = topic;
  22. }
  23. private static ConsumerConfig createConsumerConfig()
  24. {
  25. Properties props = new Properties();
  26. props.put("zookeeper.connect", KafkaProperties.zkConnect);
  27. props.put("group.id", KafkaProperties.groupId);
  28. props.put("zookeeper.session.timeout.ms", "40000");
  29. props.put("zookeeper.sync.time.ms", "200");
  30. props.put("auto.commit.interval.ms", "1000");
  31. return new ConsumerConfig(props);
  32. }
  33. @Override
  34. public void run() {
  35. Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
  36. topicCountMap.put(topic, new Integer(1));
  37. Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
  38. KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
  39. ConsumerIterator<byte[], byte[]> it = stream.iterator();
  40. while (it.hasNext()) {
  41. System.out.println("receive：" + new String(it.next().message()));
  42. try {
  43. sleep(3000);
  44. } catch (InterruptedException e) {
  45. e.printStackTrace();
  46. }
  47. }
  48. }
  49. }
简单的发送接收
 
运行下面这个程序，就可以进行简单的发送接收消息了：
  1. package com.sohu.kafkademon;
  2. /**
  3. * @author leicui bourne_cui@163.com
  4. */
  5. public class KafkaConsumerProducerDemo
  6. {
  7. public static void main(String[] args)
  8. {
  9. KafkaProducer producerThread = new KafkaProducer(KafkaProperties.topic);
  10. producerThread.start();
  11. KafkaConsumer consumerThread = new KafkaConsumer(KafkaProperties.topic);
  12. consumerThread.start();
  13. }
  14. }
高级别的consumer
 
下面是比较负载的发送接收的程序：
  1. package com.sohu.kafkademon;
  2. import java.util.HashMap;
  3. import java.util.List;
  4. import java.util.Map;
  5. import java.util.Properties;
  6. import kafka.consumer.ConsumerConfig;
  7. import kafka.consumer.ConsumerIterator;
  8. import kafka.consumer.KafkaStream;
  9. import kafka.javaapi.consumer.ConsumerConnector;
  10. /**
  11. * @author leicui bourne_cui@163.com
  12. */
  13. public class KafkaConsumer extends Thread
  14. {
  15. private final ConsumerConnector consumer;
  16. private final String topic;
  17. public KafkaConsumer(String topic)
  18. {
  19. consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
  20. createConsumerConfig());
  21. this.topic = topic;
  22. }
  23. private static ConsumerConfig createConsumerConfig()
  24. {
  25. Properties props = new Properties();
  26. props.put("zookeeper.connect", KafkaProperties.zkConnect);
  27. props.put("group.id", KafkaProperties.groupId);
  28. props.put("zookeeper.session.timeout.ms", "40000");
  29. props.put("zookeeper.sync.time.ms", "200");
  30. props.put("auto.commit.interval.ms", "1000");
  31. return new ConsumerConfig(props);
  32. }
  33. @Override
  34. public void run() {
  35. Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
  36. topicCountMap.put(topic, new Integer(1));
  37. Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
  38. KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
  39. ConsumerIterator<byte[], byte[]> it = stream.iterator();
  40. while (it.hasNext()) {
  41. System.out.println("receive：" + new String(it.next().message()));
  42. try {
  43. sleep(3000);
  44. } catch (InterruptedException e) {
  45. e.printStackTrace();
  46. }
  47. }
  48. }
  49. }
上一篇：HashMap 相关面试题及其解答


下一篇：201521123110 《Java程序设计》第1周学习总结