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

https://blog.csdn.net/qq_37171353/article/details/86533886    里面有介绍怎么安装JDK

https://blog.csdn.net/qq_37171353/article/details/80041884   里面有介绍怎么安装JDK

安装zookeeper

https://blog.csdn.net/qq_37171353/article/details/89947202  里面有介绍怎么安装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

下载地址：http://download.kafka-eagle.org/

链接：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

https://gitee.com/cbeann/Demoo/tree/master/KafkaDemo/

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 } }

 

 总结

一个分区的数据，只能交给一个消费者 一个消费者，可以消费多个分区的数据，可以跨主题 不会在同一时刻，同一组内，多个消费者同时消费数据