kafka删除topic消息的四种方式[通俗易懂]
方法一:快速配置删除法(简单粗暴,如果这个主题有程序还在消费者,此时KAFKA就game over)
1.kafka启动之前,在server.properties配置delete.topic.enable=true
2.执行命令bin/kafka-topics.sh –delete –topic test –zookeeper zk:2181或者使用kafka-manager集群管理工具删除
注意:如果kafka启动之前没有配置delete.topic.enable=true,topic只会标记为marked for deletion,加上配置,重启kafka,之前的topick就真正删除了。
方法二:设置删除策略(简单粗暴,如果这个消息有程序还在消费者,此时KAFKA就game over)
1.kafka启动之前,在server.properties配置
#日志清理策略选择有:delete和compact主要针对过期数据的处理,或是日志文件达到限制的额度,会被 topic创建时的指定参数覆盖
log.cleanup.policy = delete
# 注意:下面有两种配置,一种是基于时间的策略,另种是基于日志文件大小的策略,两种策略同是配置的话,只要满足其中种策略,则触发Log删除的操作。删除操作总是先删除最旧的日志
# 消息在Kafka中保存的时间,168小时之前的1og, 可以被删除掉,根据policy处理数据。
log.retention.hours=4
# 当剩余空间低于log.retention.bytes字节,则开始删除1og
log.retention.bytes=37580963840
# 每隔300000ms, logcleaner线程将检查一次,看是否符合上述保留策略的消息可以被删除
log.retention.check.interval.ms=1000
方法三:手动删除法(不推荐)(简单粗暴,如果这个消息有程序还在消费者,此时KAFKA就game over)
前提:不允许更改server.properties配置
1.删除zk下面topic(test)
启动bin/zkCli.sh ls /brokers/topics rmr /brokers/topics/test ls /brokers/topics 查topic是否删除:bin/kafka-topics.sh –list –zookeeper zk:2181
2.删除各broker下topic数据,默认目录为/tmp/kafka-logs
方法四:偏移量(看起来你最友好,会程序的你推荐)
package com.censoft.kafkaAdmin;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import java.sql.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
/**
* @author zy Zhang
* @version : 1.0
* @Description
* @since 2020/7/13 16:02
*/
public class DeleteReordsByOffset {
public static void main(String[] args) throws ClassNotFoundException {
// 1.创建kafkaAdminClient
Properties properties = new Properties();
properties.put("bootstrap.servers","192.168.27.111:9092");
AdminClient kafkaAdminClient = KafkaAdminClient.create(properties);
// 2.从数据库获取需要删除的消息
Class.forName("com.mysql.jdbc.Driver");
Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
String url = "jdbc:mysql://localhost:3306/test?useSSL=false&useUnicode=true&characterEncoding=UTF-8";
String user = "root";
String password = "123456";
Connection conn = null;
Statement statement = null;
ResultSet res = null;
String sql = "SELECT Topic, KafkaPartition, UntilOffset FROM Kafka_Offset;";
try {
conn = DriverManager.getConnection(url, user, password);
statement = conn.createStatement();
res = statement.executeQuery(sql);
if (res != null) {
while (res.next()) {
String topic = res.getString("Topic");
Integer partition = res.getInt("KafkaPartition");
Long offset = res.getLong("UntilOffset");
TopicPartition topicPartition = new TopicPartition(topic, partition);
RecordsToDelete recordsToDelete1 = RecordsToDelete.beforeOffset(offset);
recordsToDelete.put(topicPartition, recordsToDelete1);
}
}
} catch (SQLException e) {
e.printStackTrace();
} finally {
if (statement != null) {
try {
statement.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
// 3.执行删除
DeleteRecordsResult result = kafkaAdminClient.deleteRecords(recordsToDelete);
Map<TopicPartition, KafkaFuture<DeletedRecords>> lowWatermarks = result.lowWatermarks();
try {
for (Map.Entry<TopicPartition, KafkaFuture<DeletedRecords>> entry : lowWatermarks.entrySet()) {
System.out.println(entry.getKey().topic() + " " + entry.getKey().partition() + " " + entry.getValue().get().lowWatermark());
}
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
kafkaAdminClient.close();
}
}
2020-11-27 补充说明:
目前发现通过这种方法起到的效果是:
topic的起始偏移量会被定位到传入的recordsToDelete指定的位置
但是并没有将磁盘中存储的数据删除
如果我找到在磁盘删除的方法会继续更新,看下面
2020-11-30 补充说明: 我重新进行了测试,发现使用kafka-delete-records.sh或者kafkaAdminClient.deleteRecords()方法还有其他约束条件: 首先就是log文件自身有大小设置,对应配置文件中log.segment.bytes,在没有达到这个大小的时候是不会创建下一个log文件的。
eg: test-0下有三个log文件
00000000000000000000.log, 00000000000000000036.log, 00000000000000000136.log
我们修改起始偏移量=100
那么deleteLogStartOffsetBreachedSegments运行时会判定00000000000000000000.log是可以删除的
在原先测试时,log.segment.bytes=1G了,这造成了很难观测到数据从硬盘删除 本次测试,我将log.segment.bytes修改为了1M
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/181323.html原文链接:https://javaforall.cn
相关文章
- (3)sparkstreaming从kafka接入实时数据流最终实现数据可视化展示
- 在Golang中使用Kafka实现消息队列
- Kafka中生产消息时的三种分区分配策略
- CTO问我Pulsar到底能不能取代Kafka
- kafka学习之消息的消费原理与存储(二)
- Kafka源码解析_kafka删除消费组命令
- Kafka教程_图解kafka
- Kafka入门经典教程_kafka菜鸟教程
- elasticsearch集群搭建_Linux如何关闭kafka集群
- springboot整合kafka入门
- 如何用Know Streaming来查询Kafka的消息
- Kafka消息分区&producer拦截器&无消息丢失(八)
- Docker安装Kafka消息队列
- 面试系列-kafka消息相关机制
- KafKa主题、分区、副本、消息代理
- 不背锅运维:消息队列概念、kafka入门、Kafka Golang客户端库
- 消息队列如何选择?Kafka、Pulsar、RabbitMQ还是...
- 藏在 17 张图里的 Kafka
- Kafka集群部署文档
- kafka之一 入门介绍详解大数据
- 消息订阅之Kafka详解大数据
- Kafka 消息监控 – Kafka Eagle详解大数据
- Kafka – SQL 引擎分享详解大数据
- Kafka实战-简单示例详解大数据
- 的数据同步从MySQL到Kafka:实现实时数据同步(mysql到kafka)
- Redis与Kafka提升数据处理能力的联合组合(redis配合kafka)