前提
测试机器上需要提前实现安装好JVM，本人使用的是windows环境下利用vmare启动的Ubuntu 14.04操作系统的虚拟机，已经事先安装好了JVM 1.8。
安装实战
kafka安装包下载（注意：这里选择的安装包是2.11系列的1.1.0版本的）
wget https://archive.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz
备注：其他版本下载
安装启动服务
首先，我们需要下载并且安装zk和kafka,并且将这两个服务启动：
# 解压缩文件
tar zxvf kafka_2.11-1.1.0.tgz
cd kafka_2.11-1.1.0/
# 启动zk
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
# 检查zk是否启动成功
netstat -tunpl|grep 2181
tcp6 0 0 :::2181 :::* LISTEN 2877/java
# 启动kafka
bin/kafka-server-start.sh -daemon config/server.properties
# 检查kafka是否启动成功
netstat -tunpl|grep 9092
tcp6 0 0 :::9092 :::* LISTEN 3164/java
命令方式验证
1. 运行 producer
打开一个窗口，输入如下的指令，启动生产者：
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
2. 运行 consumer
新启动一个新的窗口，输入如下的指令，启动消费者：
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
3. 测试消息传递
在producer中写信息， 从consumer中可以看到结果, 如下：
生产者端：
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>i like you
[2018-12-07 01:55:56,427] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 1 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2018-12-07 01:55:56,532] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 3 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
>hello
>world
>hello kafka
>
消费者端：
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
i like you
hello
world
hello kafka
但是这是一种比较老使用zk启动消费端的方式，后序版本会废弃，官方推荐新的使用方式如下：
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
i like you
hello
world
hello kafka
hello
上面的消费方式为从头开始消费的，因此之前生产者发送过的历史消息，后启动的消费者仍然可以接收到。
这里我们再次测试一下只接收新的消息。
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
生产者端输入：
i like you
hello
world
hello kafka
hello
-------
louxj424 #新输入的内容
消费者端就只查看到一条消息：
louxj424
Java客户端测试代码
我们更多的时候需要在工程项目中使用kafka，下面给出一个简单的测试Demo，可以感受一下kafka的基本使用。
引入依赖的客户端jar包
首先在pom文件中引入依赖的jar包文件，注意这里一定需要安装与kafka服务端对应版本号的pom文件，必须严格对应，否则会出现各种莫名其妙的问题。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>1.1.0</version>
</dependency>
编写生产者客户端测试Demo
package com.netease.scaffold.kafka;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
public class Producer {
KafkaProducer<String, String> KafkaProducer;
public Producer() {
Properties map = new Properties();
map.put("bootstrap.servers", "192.168.91.128:9092");
map.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
map.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer = new KafkaProducer<>(map);
}
int i = 0;
String msg = "bb hh ";
public void produce() {
ProducerRecord<String, String> record = new ProducerRecord<String, String>("eagle", msg + i);
i++;
try {
KafkaProducer.send(record, (recordMetadata, e) -> System.out.println("send success"));
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
Producer producer = new Producer();
while (true) {
producer.produce();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
编写消费者客户端测试Demo
package com.netease.scaffold.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;
public class Consumer {
KafkaConsumer<String, String> kafkaConsumer;
public Consumer() {
Properties map = new Properties();
//map.put("bootstrap.servers", "59.111.60.130:9092,59.111.60.126:9092,59.111.60.127:9092");
map.put("bootstrap.servers", "192.168.91.128:9092");
map.put("group.id", "local-test-1");
map.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
map.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaConsumer = new KafkaConsumer<>(map);
kafkaConsumer.subscribe(Collections.singleton("eagle"));
}
public void consumer() {
System.out.println("wait for consume...");
try {
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.println("key = " + record.key());
System.out.println("value = " + record.value());
System.out.println("partition = " + record.partition());
System.out.println("topic = " + record.topic());
System.out.println("offset = " + record.offset());
System.out.println("timestamp = " + record.timestamp());
System.out.println();
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
Consumer consumer = new Consumer();
consumer.consumer();
}
}
功能测试
异常测试结果
踩坑经历：首次运行报错：
org.apache.kafka.common.errors.TimeoutException: Expiring 5 record(s) for test-0: 30040 ms has passe
使用java去连接kafka报连接超时错误。
解决方案：
修改kafka的config/server.properties文件中的如下内容：
advertised.listeners=PLAINTEXT://IP地址:9092
使用ifconfig或者ip a指令获取安装机器的ip地址，加入获取到的测试机器的IP地址为192.168.91.128，就将上述位置的配置参数修改为如下的内容：
advertised.listeners=PLAINTEXT://192.168.91.128:9092
修改完成后保存退出，并重新启动zk和kafka。
正常测试结果
再次运行程序。先启动消费者，消费者会先阻塞，之后再启动生产者。
- 消费者输出
key = null
value = bb hh 8
partition = 0
topic = eagle
offset = 8
timestamp = 1544183273035
key = null
value = bb hh 9
partition = 0
topic = eagle
offset = 9
timestamp = 1544183274035
key = null
value = bb hh 10
partition = 0
topic = eagle
offset = 10
timestamp = 1544183275035
key = null
value = bb hh 11
partition = 0
topic = eagle
offset = 11
timestamp = 1544183276035
- 生产者输出
send success
send success
send success
