Kafka官方给出的定义是：Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications. (Apache Kafka 是一个开源分布式事件流平台，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序。)
大数据领域中我们常用kakfa来构建流处理数据管道，与Spark或者Flink对接。
搭建Kafka集群，我们选用的kafka版本是kafka_2.12-2.4.1，Zookeeper版本为3.6.3。
zookeeper配置
# The number of milliseconds of each ticktickTime=2000# The number of ticks that the initial# synchronization phase can takeinitLimit=10# The number of ticks that can pass between# sending a request and getting an acknowledgementsyncLimit=5# the directory where the snapshot is stored.# do not use /tmp for storage, /tmp here is just# example sakes.dataDir=/tmp/zookeeper/datadataLogDir=/tmp/zookeeper/log# the port at which the clients will connectclientPort=2181# the maximum number of client connections.# increase this if you need to handle more clients#maxClientCnxns=60## Be sure to read the maintenance section of the# administrator guide before turning on autopurge.## http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance## The number of snapshots to retain in dataDir#autopurge.snapRetainCount=3# Purge task interval in hours# Set to "0" to disable auto purge feature#autopurge.purgeInterval=1## Metrics Providers## https://prometheus.io Metrics Exporter#metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider#metricsProvider.httpPort=7000#metricsProvider.exportJvmInfo=trueserver.0=10.60.**.**:2888:3888server.1=10.60.**.**:2888:3888server.2=10.60.**.**:2888:3888
默认情况下，Linux系统中没有/tmp/zookeeper/data和/tmp/zookeeper/log这两个目录，所以接下来还要创建这两个目录。
第四步，在${dataDir}目录（也就是/tmp/zookeeper/data）下创建一个myid文件，并写入一个数值，比如0。myid文件里存放的是服务器的编号。
启动
./zkServer.sh start查看zookeeper服务状态
./zkServer.sh status三个节点成功部署的话，jps后会有一个Leader和两个follower。
Kafka配置
修改 server.properties
# 指定broker的idbroker.id=0# 指定Kafka数据的位置log.dirs=/usr/local/kafka/data# 配置zk的三个节点zookeeper.connect=10.60.**.**:2181,10.60.**.**:2181,10.60.**.**:2181
将安装好的kafka复制到另外两台服务器，修改另外两个节点的broker.id分别为1和2
配置KAFKA_HOME环境变量
export KAFKA_HOME=/usr/local/kafkaexport PATH=:$PATH:${KAFKA_HOME}
启动服务器
nohup bin/kafka-server-start.sh config/server.properties &由于集群的服务器可能很多，手动启动比较麻烦，可以写一个一键启动和关闭的shell脚本：
在kafkalist文件中写入需要启动的服务器ip
cat /root/myshell/kafkalist | while read linedo{echo $linessh $line "source /etc/profile;export JMX_PORT=9988;nohup ${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties >/dev/nul* 2>&1 & "}&waitdone
示例demo
我们创建一个topic，用java代码创建一个kafka producer向topic中发送数据。
./bin/kafka-topics.sh --zookeeper 10.60.**.**:2181 --create --topic topic-demo --replication-factor 3 --partitions 4查看该topic
./bin/kafka-topics.sh --zookeeper 10.60.**.**:2181 --describe --topic topic-demoJava Producer:
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;import java.util.concurrent.ExecutionException;/**** 1. 创建用于连接Kafka的Properties配置* 2. 创建一个生产者对象KafkaProducer* 3. 调用send发送1-100消息到指定Topic test，并获取返回值Future，该对象封装了返回值* 4. 再调用一个Future.get()方法等待响应* 5. 关闭生产者*/public class KafkaProducerTest {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 创建用于连接Kafka的Properties配置Properties props = new Properties();props.put("bootstrap.servers", "117.50.**.**:9092");props.put("acks", "all");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 2. 创建一个生产者对象KafkaProducerKafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);// 3. 发送1-100的消息到指定的topic中for(int i = 0; i < 10000000; ++i) {// 一、使用同步等待的方式发送消息// // 构建一条消息，直接new ProducerRecord// ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + "");// Future<RecordMetadata> future = kafkaProducer.send(producerRecord);// // 调用Future的get方法等待响应// future.get();// System.out.println("第" + i + "条消息写入成功！");// 二、使用异步回调的方式发送消息ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic-demo", null, i + "");kafkaProducer.send(producerRecord, new Callback() {public void onCompletion(RecordMetadata metadata, Exception exception) {// 1. 判断发送消息是否成功if(exception == null) {// 发送成功// 主题String topic = metadata.topic();// 分区idint partition = metadata.partition();// 偏移量long offset = metadata.offset();System.out.println("topic:" + topic + " 分区id：" + partition + " 偏移量：" + offset);}else {// 发送出现错误System.out.println("生产消息出现异常！");// 打印异常消息System.out.println(exception.getMessage());// 打印调用栈System.out.println(exception.getStackTrace());}}});}// 4.关闭生产者kafkaProducer.close();}}
消费数据
[root kafka]# ./bin/kafka-console-consumer.sh --bootstrap-server master:9092 --topic topic-demo --from-beginning441665441666441667441668441669441670441671441672441673441674441675441676441677441678441679441680441681441682441683
可以看到数据已经写入。