Kafka笔记

CBeann

发布于 2023-12-25 17:08:52

1450

发布于 2023-12-25 17:08:52

文章被收录于专栏：CBeann的博客CBeann的博客

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）多开几个控制窗口

安装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

	}

}