（一）安装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