Kafka单机版安装
注意
1）先做一下快照
2）注意开发文章中涉及的端口
3）注意路径问题，我将软件安装在 /opt/module下，你也可以安装在 /usr/local/下
4）centos kafka_2.11-0.11.0.2.tgz zookeeper-3.4.5.tar.gz
5）iZm5ea99qngm2v98asii1aZ 是我的主机名
6）多开几个控制窗口
安装JDK
安装zookeeper
安装Kafka
解压Kafka
tar -zxvf kafka_2.11-0.11.0.2.tgz -C /opt/module/
修改 kafka_2.11-0.11.0.2/config/server.properties 文件
修改 kafka_2.11-0.11.0.2/config/producer.properties 文件
修改 kafka_2.11-0.11.0.2/config/consumer.properties文件
添加配置Kafka环境变量
注意你Kafka路径的修改
vim /etc/profile
#KAFKA_HOME export KAFKA_HOME=/opt/module/kafka export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
启动Zookeeper
bin/zkServer.sh start
启动Kafka
非后台启动
bin/kafka-server-start.sh config/server.properties
后台启动
bin/kafka-server-start.sh -daemon config/server.properties
关闭Kafka
bin/kafka-server-stop.sh stop
Kafka Eagle 监控安装
开放9999和8048端口
修改kafka-server-start.sh,添加箭头的那一行
export JMX_PORT="9999"
下载tar.gz（我们要kafka-eagle-bin-x.x.x.tar.gz中的kafka-eagle-web-1.4.1-bin.tar.gz）
下载kafka-eagle-bin-x.x.x.tar.gz并且解压，获取里面我们真正需要的kafka-eagle-web-1.4.1-bin.tar.gz
下载地址：Download - EFAK
链接：https://pan.baidu.com/s/1UKwokZKiE3ihLH-FOnDTZw
提取码：u4t3
继续解压kafka-eagle-web-1.4.1-bin.tar.gz
配置环境变量
#KE_HOME export KE_HOME=/opt/module/kafka-eagle-web-1.4.1 export PATH=$PATH:$KE_HOME/bin
赋予权限（在kafka-eagle-web目录下）
chmod 777 ke.sh
修改kafka-eagle-web-1.4.1/conf/system-config.properties配置文件
下面可以配，也可以不配
###################################### # kafka mysql jdbc driver address ###################################### kafka.eagle.driver=com.mysql.jdbc.Driver kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull kafka.eagle.username=root kafka.eagle.password=root
启动监控插件
确保zk和kafka都启动
启动监控插件
./bin/ke.sh start
关闭监控插件
./bin/ke.sh stop
下图表示安装并且开启成功
遇见的问题
问题：正常启动但是立马关闭
问题原因：内存不足
解决办法：暂时还没找到，好菜
Kafka 命令行操作
查看当前服务器中的所有 topic
bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --list
创建 topic
选项说明： --topic 定义 topic 名 --replication-factor 定义副本数 --partitions 定义分区数
bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --create --replication-factor 1 --partitions 3 --topic first
删除 topic
需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除或者直接重启。
bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --delete --topic first
启动控制台生产消息
bin/kafka-console-producer.sh --broker-list iZm5ea99qngm2v98asii1aZ:9092 --topic first
启动控制台消费消息
--from-beginning：会把 first 主题中以往所有的数据都读取出来。根据业务场景选择是否 增加该配置。
bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first
开启多个消费者位于同一个消费者组中
1）启动zookeeper
bin/zkServer.sh start
2) 启动Kafka
非后台启动
bin/kafka-server-start.sh config/server.properties
后台启动
bin/kafka-server-start.sh -daemon config/server.properties
3) 创建 名称为 first 的topic
bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --create --replication-factor 1 --partitions 3 --topic first
4）像first中生产消息
bin/kafka-console-producer.sh --broker-list iZm5ea99qngm2v98asii1aZ:9092 --topic first
5）两个命令行消费消息（重点）（重点）（重点）
bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first --consumer.config config/consumer.properties
6）运行结果：一个消费者组中同一时刻只能有一个消费者消费
KafkaAPI
Demoo: SSM,SpringBoot or other demo - Gitee.com
pom.xml
依赖的版本要和
<dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>3.8.1</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.2</version> </dependency>
生产者
不带回调函数的生产者
package com.kafka.producer; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; /** * 创建生成者：生产消息 * 1）启动zookeeper * ./bin/zkServer.sh start * 2)启动Kafka server * bin/kafka-server-start.sh config/server.properties * 3)启动控制台消息消费者 * bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first * 4)运行Main方法 * */ public class CustomProducer { public static void main(String[] args) { Properties props = new Properties(); // Kafka服务端的主机名和端口号 //props.put("bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092"); props.put("bootstrap.servers", "iZm5ea99qngm2v98asii1aZ:9092"); // 等待所有副本节点的应答 props.put("acks", "all"); // 消息发送最大尝试次数 props.put("retries", 0); // 一批消息处理大小 props.put("batch.size", 16384); // 请求延时 props.put("linger.ms", 1); // 发送缓存区内存大小 props.put("buffer.memory", 33554432); // key序列化 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //实例化KafkaProducer KafkaProducer<String, String> producer = new KafkaProducer<>(props); for (int i = 0; i < 50; i++) { //public ProducerRecord(String topic, K key, V value) //发送消息 producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), "hello world-" + i)); } //释放资源 producer.close(); System.out.println("-------------OVER-------------------"); } }
带有回调函数的生产者
package com.kafka.producer; import java.util.Properties; 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; /** * 创建生成者：生产消息 * 1）启动zookeeper * ./bin/zkServer.sh start * 2)启动Kafka server * bin/kafka-server-start.sh config/server.properties * 3)启动控制台消息消费者 * bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first * 4)运行Main方法 * */ public class CustomProducerWithCallBack { public static void main(String[] args) throws InterruptedException { Properties props = new Properties(); // Kafka服务端的主机名和端口号 props.put("bootstrap.servers", "iZm5ea99qngm2v98asii1aZ:9092"); // 等待所有副本节点的应答 props.put("acks", "all"); // 消息发送最大尝试次数 props.put("retries", 0); // 一批消息处理大小 props.put("batch.size", 16384); // 增加服务端请求延时 props.put("linger.ms", 1); // 发送缓存区内存大小 props.put("buffer.memory", 33554432); // key序列化 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //实例化KafkaProducer KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props); for (int i = 0; i < 50; i++) { //发送消息 kafkaProducer.send(new ProducerRecord<String, String>("test1", "hh" + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (metadata != null) { System.out.println(metadata.partition() + "---" + metadata.offset()); } } }); } kafkaProducer.close(); } }
自定义分区
package com.kafka.producer; import java.util.Map; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.common.Cluster; // 自定义分区 //props.put("partitioner.class", "com.kafka.CustomPartitioner"); public class CustomPartitioner implements Partitioner { @Override public void configure(Map<String, ?> configs) { } @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { // 控制分区 //巴里巴拉 return 1; } @Override public void close() { } }
消费者
package com.kafka.consumer; 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; /** * 创建消费者：消费消息 * 1）启动zookeeper * ./bin/zkServer.sh start * 2)启动Kafka server * bin/kafka-server-start.sh config/server.properties * 3)启动控制台生产消息 * bin/kafka-console-producer.sh --broker-list iZm5ea99qngm2v98asii1aZ:9092 --topic first * 4)运行Main方法 * */ public class CustomConsumer { public static void main(String[] args) { Properties props = new Properties(); // 定义kakfa 服务的地址 props.put("bootstrap.servers", "iZm5ea99qngm2v98asii1aZ:9092"); // 制定consumer group props.put("group.id", "g1"); // 是否自动确认offset props.put("enable.auto.commit", "true"); // 自动确认offset的时间间隔 props.put("auto.commit.interval.ms", "1000"); // key的序列化类 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // value的序列化类 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 定义consumer KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); // 消费者订阅的topic, 可同时订阅多个 consumer.subscribe(Arrays.asList("first")); //拉取消息 while (true) { // 读取数据，读取超时时间为100ms ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } } }
拦截器：实现ProducerInterceptor接口
package com.kafka.interceptor; import java.util.Map; import org.apache.kafka.clients.producer.ProducerInterceptor; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; /** * 通过拦截器添加时间戳 * */ // 在main 方法中添加过滤器 // List<String> interceptors = new ArrayList<>(); // interceptors.add("com.kafka.interceptor.TimeInterceptor"); // props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors); // String topic = "first"; // Producer<String, String> producer = new KafkaProducer<>(props); public class TimeInterceptor implements ProducerInterceptor<String, String> { @Override public void configure(Map<String, ?> configs) { // TODO Auto-generated method stub } @Override public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { // 创建一个新的record，把时间戳写入消息体的最前部 return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(), System.currentTimeMillis() + "," + record.value().toString()); } @Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) { // TODO Auto-generated method stub } @Override public void close() { // TODO Auto-generated method stub } }
总结
一个分区的数据，只能交给一个消费者
一个消费者，可以消费多个分区的数据，可以跨主题
不会在同一时刻，同一组内，多个消费者同时消费数据
