kafka环境搭建（WindowsLinux）

（一）安装zookeeper(windows)

kafka需要用到zookeeper,所以需要先安装zookeeper

1.到官网下载最新版zookeeper，http://www.apache.org/dyn/closer.cgi/zookeeper/

2.解压到你喜欢的路径，我这里为：E:\zookeeper\zookeeper-3.4.10

3.复制conf目录下zoo_sample.cfg，粘贴改名为zoo.cfg，修改zoo.cfg中的dataDir的值为E:/data/zookeeper，并添加一行dataLogDir=E:/log/zookeeper

4.修改系统环境变量，在Path后添加    ;E:\zookeeper\zookeeper-3.4.10\bin

5.运行cmd命令窗口，输入zkServer回车，出现下图的就表示zookeeper启动成功，也表明安装成功了。

安装zookeeper(Linux)

1. Xshell等工具连接Linux服务器，切换到任意目录，下载zookeeper最新稳定版，下载地址http://mirrors.hust.edu.cn/apache/zookeeper/stable/，命令如下

cd /usr/soft

wget http://mirrors.hust.edu.cn/apache/zookeeper/stable/zookeeper-3.4.10.tar.gz

2.解压

tar -xzvf zookeeper-3.4.10.tar.gz

3.切换到conf配置文件目录，复制zoo_sample.cfg为zoo.cfg可以按需修改配置文件内容

4.切换到bin目录,启动zookeeper,看到Starting zookeeper ... STARTED字样表示启动成功了

./zkServer.sh start

（二）安装kafka(windows)

1. 到官网下载最新版kafka，http://kafka.apache.org/downloads

2.解压到你喜欢的路径，我这里解压路径为：E:\kafka_2.12-0.10.2.0

3.修改E:\kafka_2.12-0.10.2.0\config目录下的server.properties中 log.dirs的值为E:/log/kafka

4.添加系统环境变量，在Path后添加   ;E:\kafka_2.12-0.10.2.0\bin\windows

5.启动kafka，在cmd命令行用cd命令切换到kafka根目录E:\kafka_2.12-0.10.2.0，输入命令

 .\bin\windows\kafka-server-start.bat .\config\server.properties

出现started (kafka.server.KafkaServer)字样表示启动成功

启动时若出现“wvim不是内部或外部命令...”错误提示，则需要在系统Path环境变量后添加 ;C:\Windows\System32\wbem 6.运行cmd命令行，创建一个topic

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

7.再打开一个cmd，创建一个Producer

kafka-console-producer.bat --broker-list localhost:9092 --topic test

8.再打开一个cmd，创建一个Customer

kafka-console-consumer.bat --zookeeper localhost:2181 --topic test

9.在Producer窗口下输入信息进行测试 ，每输入一行回车后消息马上就会出现在Customer中，表明kafka已经安装测试成功

安装kafka(Linux)

1.下载kafka最新版https://www.apache.org/dyn/closer.cgi?path=/kafka/0.10.2.0/kafka_2.12-0.10.2.0.tgz

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.10.2.0/kafka_2.12-0.10.2.0.tgz

2.解压,文件夹重命名

tar -xzvf kafka_2.12-0.10.2.0.tgz

mv kafka_2.12-0.10.2.0 kafka

3.切换目录到kafka目录下的bin目录，用vi命令修改kafka-server-start.sh中jvm内存大小，把

export KAFKA_HEAP_OPTS="-Xms1G -Xms1G" 修改为 export KAFKA_HEAP_OPTS="-Xms256M -Xms128M"，当然如果你的内存够大可以不修改

4.切换到kafka根目录，启动kafka，启动成功如下图

bin/kafka-server-start.sh config/server.properties

5.创建topic bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic test 创建一个名为test的topic，只有一个副本，一个分区。 通过list命令查看刚刚创建的topic bin/kafka-topics.sh -list -zookeeper 127.0.0.1:2181 6.启动producer并发送消息启动producer bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 启动之后就可以发送消息了 按Ctrl+C退出发送消息 7.启动consumer bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning 启动consumer之后就可以在console中看到producer发送的消息了 可以开启两个终端，一个发送消息，一个接受消息。

（三）kafka编程之Java接口

1.新建Maven工程，我这里用的是Eclipse；pom加入kafka依赖，如下：

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.10.2.0</version> </dependency> 2.新建生产测试类TestProducer.java

import java.util.Properties; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; public class TestProducer { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); //The "all" setting we have specified will result in blocking on the full commit of the record, the slowest but most durable setting. //“所有”设置将导致记录的完整提交阻塞，最慢的，但最持久的设置。 props.put("acks", "all"); //如果请求失败，生产者也会自动重试，即使设置成０ the producer can automatically retry. props.put("retries", 0); //The producer maintains buffers of unsent records for each partition. props.put("batch.size", 16384); //默认立即发送，这里这是延时毫秒数 props.put("linger.ms", 1); //生产者缓冲大小，当缓冲区耗尽后，额外的发送调用将被阻塞。时间超过max.block.ms将抛出TimeoutException props.put("buffer.memory", 33554432); //The key.serializer and value.serializer instruct how to turn the key and value objects the user provides with their ProducerRecord into bytes. props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //创建kafka的生产者类 Producer<String, String> producer = new KafkaProducer<String, String>(props); //生产者的主要方法 // close();//Close this producer. // close(long timeout, TimeUnit timeUnit); //This method waits up to timeout for the producer to complete the sending of all incomplete requests. // flush() ;所有缓存记录被立刻发送 for(int i = 0; i < 100; i++) producer.send(new ProducerRecord<String, String>("test",0, Integer.toString(i), Integer.toString(i))); producer.close(); } } 3.新建消费测试类TestCustomer.java

import java.util.Arrays; import java.util.Properties; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; public class TestConsumer { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); System.out.println("this is the group part test 1"); //消费者的组id props.put("group.id", "GroupA");//这里是GroupA或者GroupB props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); //从poll(拉)的回话处理时长 props.put("session.timeout.ms", "30000"); //poll的数量限制 //props.put("max.poll.records", "100"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); //订阅主题列表topic consumer.subscribe(Arrays.asList("test")); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) //　正常这里应该使用线程池处理，不应该在这里处理 System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value()+"\n"); } } } 4.先运行（run/debug）TestCustomer， 再运行TestProducer，在TestCustomer的控制台看到下图的结果就表示消息发送并接收成功了

并且在之前启动的消费端的命令窗口也能看到接收到的数据：

dazu表示kakazhj