提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  数据库

当前栏目

Kafka笔记

Kafka 笔记
2023-03-14 22:45:21 时间

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 文件


1.png


2.png


修改 kafka_2.11-0.11.0.2/config/producer.properties 文件


3.png


修改 kafka_2.11-0.11.0.2/config/consumer.properties文件


4.png


添加配置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"


5.png


下载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


6.png


修改kafka-eagle-web-1.4.1/conf/system-config.properties配置文件


7.png


8.png


9.png


下面可以配，也可以不配


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


10.png


 启动监控插件


确保zk和kafka都启动

启动监控插件


./bin/ke.sh start


11.png


关闭监控插件


./bin/ke.sh stop


下图表示安装并且开启成功


12.png


遇见的问题


问题：正常启动但是立马关闭

问题原因：内存不足


13.png


解决办法：暂时还没找到，好菜 


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


14.png


删除 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


15.png


启动控制台消费消息


--from-beginning：会把 first 主题中以往所有的数据都读取出来。根据业务场景选择是否 增加该配置。


bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first


16.png


开启多个消费者位于同一个消费者组中


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）运行结果：一个消费者组中同一时刻只能有一个消费者消费


17.png


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


 总结


一个分区的数据，只能交给一个消费者

一个消费者，可以消费多个分区的数据，可以跨主题

不会在同一时刻，同一组内，多个消费者同时消费数据

相关文章