（转）Kafka部署与代码实例

标签： kafka 代码 实例 | 发表时间：2014-08-14 13:43 | 作者：kavy
出处：http://www.iteye.com

转自： http://shift-alt-ctrl.iteye.com/blog/1930791

  kafka作为分布式日志收集或系统监控服务，我们有必要在合适的场合使用它。kafka的部署包括zookeeper环境/kafka环境，同时还需要进行一些配置操作.接下来介绍如何使用kafka.

    我们使用3个zookeeper实例构建zk集群，使用2个kafka broker构建kafka集群.

    其中kafka为0.8V，zookeeper为3.4.5V

 

一.Zookeeper集群构建

    我们有3个zk实例，分别为zk-0,zk-1,zk-2;如果你仅仅是测试使用，可以使用1个zk实例.

     1) zk-0

    调整配置文件：

Php代码   收藏代码
  1. clientPort=2181  
  2. server.0=127.0.0.1:2888:3888  
  3. server.1=127.0.0.1:2889:3889  
  4. server.2=127.0.0.1:2890:3890  
  5. ##只需要修改上述配置，其他配置保留默认值  

    启动zookeeper

Java代码   收藏代码
  1. ./zkServer.sh start  

     2) zk-1

    调整配置文件(其他配置和zk-0一只)：

Php代码   收藏代码
  1. clientPort=2182  
  2. ##只需要修改上述配置，其他配置保留默认值  

    启动zookeeper

 

Java代码   收藏代码
  1. ./zkServer.sh start  

     3) zk-2

    调整配置文件(其他配置和zk-0一只)：

Php代码   收藏代码
  1. clientPort=2183  
  2. ##只需要修改上述配置，其他配置保留默认值  

    启动zookeeper

 

Java代码   收藏代码
  1. ./zkServer.sh start  

  

二. Kafka集群构建

    因为Broker配置文件涉及到zookeeper的相关约定，因此我们先展示broker配置文件.我们使用2个kafka broker来构建这个集群环境，分别为kafka-0,kafka-1.

     1) kafka-0

    在config目录下修改配置文件为：

Java代码   收藏代码
  1. broker.id=0  
  2. port=9092  
  3. num.network.threads=2  
  4. num.io.threads=2  
  5. socket.send.buffer.bytes=1048576  
  6. socket.receive.buffer.bytes=1048576  
  7. socket.request.max.bytes=104857600  
  8. log.dir=./logs  
  9. num.partitions=2  
  10. log.flush.interval.messages=10000  
  11. log.flush.interval.ms=1000  
  12. log.retention.hours=168  
  13. #log.retention.bytes=1073741824  
  14. log.segment.bytes=536870912  
  15. ##replication机制,让每个topic的partitions在kafka-cluster中备份2个  
  16. ##用来提高cluster的容错能力..  
  17. default.replication.factor=1  
  18. log.cleanup.interval.mins=10  
  19. zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183  
  20. zookeeper.connection.timeout.ms=1000000  

    因为kafka用scala语言编写，因此运行kafka需要首先准备scala相关环境。

Java代码   收藏代码
  1. > cd kafka-0  
  2. > ./sbt update  
  3. > ./sbt package  
  4. > ./sbt assembly-package-dependency   

    其中最后一条指令执行有可能出现异常，暂且不管。 启动kafka broker：

Java代码   收藏代码
  1. > JMS_PORT=9997 bin/kafka-server-start.sh config/server.properties &  

    因为zookeeper环境已经正常运行了，我们无需通过kafka来挂载启动zookeeper.如果你的一台机器上部署了多个kafka broker，你需要声明JMS_PORT.

     2) kafka-1

Java代码   收藏代码
  1. broker.id=1  
  2. port=9093  
  3. ##其他配置和kafka-0保持一致  

    然后和kafka-0一样执行打包命令，然后启动此broker.

Java代码   收藏代码
  1. > JMS_PORT=9998 bin/kafka-server-start.sh config/server.properties &  

    仍然可以通过如下指令查看topic的"partition"/"replicas"的分布和存活情况.

Java代码   收藏代码
  1. > bin/kafka-list-topic.sh --zookeeper localhost:2181  
  2. topic: my-replicated-topic  partition: 0    leader: 2   replicas: 1,2,0 isr: 2  
  3. topic: test partition: 0    leader: 0   replicas: 0 isr: 0   

    到目前为止环境已经OK了,那我们就开始展示编程实例吧。[ 配置参数详解]

 

三.项目准备

    项目基于maven构建，不得不说kafka java客户端实在是太糟糕了；构建环境会遇到很多麻烦。建议参考如下pom.xml;其中各个依赖包必须版本协调一致。如果kafka client的版本和kafka server的版本不一致,将会有很多异常,比如"broker id not exists"等;因为kafka从0.7升级到0.8之后(正名为2.8.0),client与server通讯的protocol已经改变.

Java代码   收藏代码
  1. <dependencies>  
  2.     <dependency>  
  3.         <groupId>log4j</groupId>  
  4.         <artifactId>log4j</artifactId>  
  5.         <version>1.2.14</version>  
  6.     </dependency>  
  7.     <dependency>  
  8.         <groupId>org.apache.kafka</groupId>  
  9.         <artifactId>kafka_2.8.2</artifactId>  
  10.         <version>0.8.0</version>  
  11.         <exclusions>  
  12.             <exclusion>  
  13.                 <groupId>log4j</groupId>  
  14.                 <artifactId>log4j</artifactId>  
  15.             </exclusion>  
  16.         </exclusions>  
  17.     </dependency>  
  18.     <dependency>  
  19.         <groupId>org.scala-lang</groupId>  
  20.         <artifactId>scala-library</artifactId>  
  21.         <version>2.8.2</version>  
  22.     </dependency>  
  23.     <dependency>  
  24.         <groupId>com.yammer.metrics</groupId>  
  25.         <artifactId>metrics-core</artifactId>  
  26.         <version>2.2.0</version>  
  27.     </dependency>  
  28.     <dependency>  
  29.         <groupId>com.101tec</groupId>  
  30.         <artifactId>zkclient</artifactId>  
  31.         <version>0.3</version>  
  32.     </dependency>  
  33. </dependencies>  

 

四.Producer端代码

     1) producer.properties文件：此文件放在/resources目录下

Java代码   收藏代码
  1. #partitioner.class=  
  2. ##broker列表可以为kafka server的子集,因为producer需要从broker中获取metadata  
  3. ##尽管每个broker都可以提供metadata,此处还是建议,将所有broker都列举出来  
  4. ##此值,我们可以在spring中注入过来  
  5. ##metadata.broker.list=127.0.0.1:9092,127.0.0.1:9093  
  6. ##,127.0.0.1:9093  
  7. ##同步,建议为async  
  8. producer.type=sync  
  9. compression.codec=0  
  10. serializer.class=kafka.serializer.StringEncoder  
  11. ##在producer.type=async时有效  
  12. #batch.num.messages=100  

     2) KafkaProducerClient.java代码样例

Java代码   收藏代码
  1. import java.util.ArrayList;  
  2. import java.util.Collection;  
  3. import java.util.List;  
  4. import java.util.Properties;  
  5.   
  6. import kafka.javaapi.producer.Producer;  
  7. import kafka.producer.KeyedMessage;  
  8. import kafka.producer.ProducerConfig;  
  9.   
  10. /** 
  11.  * User: guanqing-liu 
  12.  */  
  13. public class KafkaProducerClient {  
  14.   
  15.     private Producer<String, String> inner;  
  16.       
  17.     private String brokerList;//for metadata discovery,spring setter  
  18.     private String location = "kafka-producer.properties";//spring setter  
  19.       
  20.     private String defaultTopic;//spring setter  
  21.   
  22.     public void setBrokerList(String brokerList) {  
  23.         this.brokerList = brokerList;  
  24.     }  
  25.   
  26.     public void setLocation(String location) {  
  27.         this.location = location;  
  28.     }  
  29.   
  30.     public void setDefaultTopic(String defaultTopic) {  
  31.         this.defaultTopic = defaultTopic;  
  32.     }  
  33.   
  34.     public KafkaProducerClient(){}  
  35.       
  36.     public void init() throws Exception {  
  37.         Properties properties = new Properties();  
  38.         properties.load(Thread.currentThread().getContextClassLoader().getResourceAsStream(location));  
  39.           
  40.           
  41.         if(brokerList != null) {  
  42.             properties.put("metadata.broker.list", brokerList);  
  43.         }  
  44.   
  45.         ProducerConfig config = new ProducerConfig(properties);  
  46.         inner = new Producer<String, String>(config);  
  47.     }  
  48.   
  49.     public void send(String message){  
  50.         send(defaultTopic,message);  
  51.     }  
  52.       
  53.     public void send(Collection<String> messages){  
  54.         send(defaultTopic,messages);  
  55.     }  
  56.       
  57.     public void send(String topicName, String message) {  
  58.         if (topicName == null || message == null) {  
  59.             return;  
  60.         }  
  61.         KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message);  
  62.         inner.send(km);  
  63.     }  
  64.   
  65.     public void send(String topicName, Collection<String> messages) {  
  66.         if (topicName == null || messages == null) {  
  67.             return;  
  68.         }  
  69.         if (messages.isEmpty()) {  
  70.             return;  
  71.         }  
  72.         List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>();  
  73.         int i= 0;  
  74.         for (String entry : messages) {  
  75.             KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,entry);  
  76.             kms.add(km);  
  77.             i++;  
  78.             if(i % 20 == 0){  
  79.                 inner.send(kms);  
  80.                 kms.clear();  
  81.             }  
  82.         }  
  83.           
  84.         if(!kms.isEmpty()){  
  85.             inner.send(kms);  
  86.         }  
  87.     }  
  88.   
  89.     public void close() {  
  90.         inner.close();  
  91.     }  
  92.   
  93.     /** 
  94.      * @param args 
  95.      */  
  96.     public static void main(String[] args) {  
  97.         KafkaProducerClient producer = null;  
  98.         try {  
  99.             producer = new KafkaProducerClient();  
  100.             //producer.setBrokerList("");  
  101.             int i = 0;  
  102.             while (true) {  
  103.                 producer.send("test-topic", "this is a sample" + i);  
  104.                 i++;  
  105.                 Thread.sleep(2000);  
  106.             }  
  107.         } catch (Exception e) {  
  108.             e.printStackTrace();  
  109.         } finally {  
  110.             if (producer != null) {  
  111.                 producer.close();  
  112.             }  
  113.         }  
  114.   
  115.     }  
  116.   
  117. }  

     3) spring配置

Java代码   收藏代码
  1. <bean id="kafkaProducerClient" class="com.test.kafka.KafkaProducerClient" init-method="init" destroy-method="close">  
  2.     <property name="zkConnect" value="${zookeeper_cluster}"></property>  
  3.     <property name="defaultTopic" value="${kafka_topic}"></property>  
  4. </bean>  

 

五.Consumer端

     1) consumer.properties:文件位于/resources目录下

Java代码   收藏代码
  1. ## 此值可以配置,也可以通过spring注入  
  2. ##zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183  
  3. ##,127.0.0.1:2182,127.0.0.1:2183  
  4. # timeout in ms for connecting to zookeeper  
  5. zookeeper.connectiontimeout.ms=1000000  
  6. #consumer group id  
  7. group.id=test-group  
  8. #consumer timeout  
  9. #consumer.timeout.ms=5000  
  10. auto.commit.enable=true  
  11. auto.commit.interval.ms=60000  

     2) KafkaConsumerClient.java代码样例

Java代码   收藏代码
  1. package com.test.kafka;  
  2. import java.nio.ByteBuffer;  
  3. import java.nio.CharBuffer;  
  4. import java.nio.charset.Charset;  
  5. import java.util.HashMap;  
  6. import java.util.List;  
  7. import java.util.Map;  
  8. import java.util.Properties;  
  9. import java.util.concurrent.ExecutorService;  
  10. import java.util.concurrent.Executors;  
  11.   
  12. import kafka.consumer.Consumer;  
  13. import kafka.consumer.ConsumerConfig;  
  14. import kafka.consumer.ConsumerIterator;  
  15. import kafka.consumer.KafkaStream;  
  16. import kafka.javaapi.consumer.ConsumerConnector;  
  17. import kafka.message.Message;  
  18. import kafka.message.MessageAndMetadata;  
  19.   
  20. /** 
  21.  * User: guanqing-liu  
  22.  */  
  23. public class KafkaConsumerClient {  
  24.   
  25.     private String groupid; //can be setting by spring  
  26.     private String zkConnect;//can be setting by spring  
  27.     private String location = "kafka-consumer.properties";//配置文件位置  
  28.     private String topic;  
  29.     private int partitionsNum = 1;  
  30.     private MessageExecutor executor; //message listener  
  31.     private ExecutorService threadPool;  
  32.       
  33.     private ConsumerConnector connector;  
  34.       
  35.     private Charset charset = Charset.forName("utf8");  
  36.   
  37.     public void setGroupid(String groupid) {  
  38.         this.groupid = groupid;  
  39.     }  
  40.   
  41.     public void setZkConnect(String zkConnect) {  
  42.         this.zkConnect = zkConnect;  
  43.     }  
  44.   
  45.     public void setLocation(String location) {  
  46.         this.location = location;  
  47.     }  
  48.   
  49.     public void setTopic(String topic) {  
  50.         this.topic = topic;  
  51.     }  
  52.   
  53.     public void setPartitionsNum(int partitionsNum) {  
  54.         this.partitionsNum = partitionsNum;  
  55.     }  
  56.   
  57.     public void setExecutor(MessageExecutor executor) {  
  58.         this.executor = executor;  
  59.     }  
  60.   
  61.     public KafkaConsumerClient() {}  
  62.   
  63.     //init consumer,and start connection and listener  
  64.     public void init() throws Exception {  
  65.         if(executor == null){  
  66.             throw new RuntimeException("KafkaConsumer,exectuor cant be null!");  
  67.         }  
  68.         Properties properties = new Properties();  
  69.         properties.load(Thread.currentThread().getContextClassLoader().getResourceAsStream(location));  
  70.           
  71.         if(groupid != null){  
  72.             properties.put("groupid", groupid);  
  73.         }  
  74.         if(zkConnect != null){  
  75.             properties.put("zookeeper.connect", zkConnect);  
  76.         }  
  77.         ConsumerConfig config = new ConsumerConfig(properties);  
  78.   
  79.         connector = Consumer.createJavaConsumerConnector(config);  
  80.         Map<String, Integer> topics = new HashMap<String, Integer>();  
  81.         topics.put(topic, partitionsNum);  
  82.         Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(topics);  
  83.         List<KafkaStream<byte[], byte[]>> partitions = streams.get(topic);  
  84.         threadPool = Executors.newFixedThreadPool(partitionsNum * 2);  
  85.           
  86.         //start  
  87.         for (KafkaStream<byte[], byte[]> partition : partitions) {  
  88.             threadPool.execute(new MessageRunner(partition));  
  89.         }  
  90.     }  
  91.   
  92.     public void close() {  
  93.         try {  
  94.             threadPool.shutdownNow();  
  95.         } catch (Exception e) {  
  96.             //  
  97.         } finally {  
  98.             connector.shutdown();  
  99.         }  
  100.   
  101.     }  
  102.   
  103.     class MessageRunner implements Runnable {  
  104.         private KafkaStream<byte[], byte[]> partition;  
  105.   
  106.         MessageRunner(KafkaStream<byte[], byte[]> partition) {  
  107.             this.partition = partition;  
  108.         }  
  109.   
  110.         public void run() {  
  111.             ConsumerIterator<byte[], byte[]> it = partition.iterator();  
  112.             while (it.hasNext()) {  
  113.                 // connector.commitOffsets();手动提交offset,当autocommit.enable=false时使用  
  114.                 MessageAndMetadata<byte[], byte[]> item = it.next();  
  115.                 try{  
  116.                     executor.execute(new String(item.message(),charset));// UTF-8,注意异常  
  117.                 }catch(Exception e){  
  118.                     //  
  119.                 }  
  120.             }  
  121.         }  
  122.           
  123.         public String getContent(Message message){  
  124.             ByteBuffer buffer = message.payload();  
  125.             if (buffer.remaining() == 0) {  
  126.                 return null;  
  127.             }  
  128.             CharBuffer charBuffer = charset.decode(buffer);  
  129.             return charBuffer.toString();  
  130.         }  
  131.     }  
  132.   
  133.     public static interface MessageExecutor {  
  134.   
  135.         public void execute(String message);  
  136.     }  
  137.   
  138.     /** 
  139.      * @param args 
  140.      */  
  141.     public static void main(String[] args) {  
  142.         KafkaConsumerClient consumer = null;  
  143.         try {  
  144.             MessageExecutor executor = new MessageExecutor() {  
  145.   
  146.                 public void execute(String message) {  
  147.                     System.out.println(message);  
  148.                 }  
  149.             };  
  150.             consumer = new KafkaConsumerClient();  
  151.               
  152.             consumer.setTopic("test-topic");  
  153.             consumer.setPartitionsNum(2);  
  154.             consumer.setExecutor(executor);  
  155.             consumer.init();  
  156.         } catch (Exception e) {  
  157.             e.printStackTrace();  
  158.         } finally {  
  159.              if(consumer != null){  
  160.                  consumer.close();  
  161.              }  
  162.         }  
  163.   
  164.     }  
  165.   
  166. }  

    3) spring配置(略)

 

    需要提醒的是,上述LogConsumer类中,没有太多的关注异常情况,必须在MessageExecutor.execute()方法中抛出异常时的情况.

    在测试时，建议优先启动consumer，然后再启动producer，这样可以实时的观测到最新的消息。



已有 0 人发表留言，猛击->> 这里<<-参与讨论


ITeye推荐



相关 [kafka 代码 实例] 推荐：

（转）Kafka部署与代码实例

- - 开源软件 - ITeye博客
转自： http://shift-alt-ctrl.iteye.com/blog/1930791.   kafka作为分布式日志收集或系统监控服务，我们有必要在合适的场合使用它. kafka的部署包括zookeeper环境/kafka环境，同时还需要进行一些配置操作.接下来介绍如何使用kafka..

Kafka编程实例

- - CSDN博客云计算推荐文章
    Producer是一个应用程序，它创建消息并发送它们到Kafka broker中. 这些producer在本质上是不同. 比如，前端应用程序，后端服务，代理服务，适配器对于潜在的系统，Hadoop对于的Producer. 这些不同的Producer能够使用不同的语言实现，比如java、C和Python.

kafka开发实例

- - 互联网 - ITeye博客
//启动zookeeper server (用&是为了能退出命令行)：. //启动kafka server: . 已有 0 人发表留言，猛击->> 这里<<-参与讨论. —软件人才免语言低担保 赴美带薪读研.

kafka监控之kafka-run-class.sh

- - 开源软件 - ITeye博客
kafka自带了很多工具类，在源码kafka.tools里可以看到：. 这些类该如何使用呢，kafka的设计者早就为我们考虑到了，在${KAFKA_HOME}/bin下，有很多的脚本，其中有一个kafka-run-class.sh，通过这个脚本，可以调用其中的tools的部分功能，如调用kafka.tools里的ConsumerOffsetChecker.scala,.

闲扯kafka mq

- - 开源软件 - ITeye博客
本文主要讲解关于kafka mq的设计思想及个人理解. 关于kafka的详细信息，大家可以参考官网的文献 http://kafka.apache.org/documentation.html这是一篇相当不错的文章，值得仔细研读. 第一个问题：消息队列（Message Queue）是干嘛用的. 首先，要对消息队列有一个基本的理解.

Kafka优化

- - ITeye博客
配置优化都是修改server.properties文件中参数值. 1.网络和io操作线程配置优化. # broker处理消息的最大线程数. # broker处理磁盘IO的线程数. 一般num.network.threads主要处理网络io，读写缓冲区数据，基本没有io等待，配置线程数量为cpu核数加1.

Kafka Connect简介

- - 鸟窝
Kafka 0.9+增加了一个新的特性 Kafka Connect,可以更方便的创建和管理数据流管道. 它为Kafka和其它系统创建规模可扩展的、可信赖的流数据提供了一个简单的模型，通过 connectors可以将大数据从其它系统导入到Kafka中，也可以从Kafka中导出到其它系统. Kafka Connect可以将完整的数据库注入到Kafka的Topic中，或者将服务器的系统监控指标注入到Kafka，然后像正常的Kafka流处理机制一样进行数据流处理.

kafka consumer group offset

- - 开源软件 - ITeye博客
     kafka0.9及以前版本kafka offset 保存在zookeeper， 因频繁读写zookeeper性能不高；从0.10开始，主题分区offset存储于kafka独立主题中.     管理监控kafka主题及分区offset至关重要，原网上很开源流行工具KafkaOffsetMonitor、kafka-manager，旧版offset保存于zookeeper，kafka consumer无相应API，从kafka0.10.1.1以后提供相应API读取主题分区offset（也可以调用KafkaClient API，kafka管理API由scala语言编写）.

Kafka设计解析（二）：Kafka High Availability （上）

- -
Kafka在0.8以前的版本中，并不提供High Availablity机制，一旦一个或多个Broker宕机，则宕机期间其上所有Partition都无法继续提供服务. 若该Broker永远不能再恢复，亦或磁盘故障，则其上数据将丢失. 而Kafka的设计目标之一即是提供数据持久化，同时对于分布式系统来说，尤其当集群规模上升到一定程度后，一台或者多台机器宕机的可能性大大提高，对Failover要求非常高.

GitHub - andreas-schroeder/kafka-health-check: Health Check for Kafka Brokers.

- -
At AutoScout24, to keep the OS up to date of our clusters running on AWS, we perform regular in-place rolling updates. As we run immutable servers, we terminate each broker and replace them with fresh EC2 instances (keeping the previous broker ids).