一、添加依赖org.apache.kafkakafka-clients3.5.1二、生产者自定义分区，可忽略importorg.apache.kafka.clients.producer.Partitioner;importorg.apache.kafka.common.Cluster;importjava.util.Map;publicclassMyPatitionerimplementsPartitioner{@Overridepublicintpartition(Stringtopic,Objectkey,byte[]keyBytes,Objectvalue,byte[]valueByt
Kafka安装Kafka安装包下载地址：https://archive.apache.org/dist/kafka/1.Kafka伪分布式安装1.上传并解压Kafka安装包使用FileZilla或其他文件传输工具上传Kafka安装包：kafka_2.11-0.10.0.0.tgz解压安装包[root@bigdatasoftware]#tar-zxvfkafka_2.11-0.10.0.0.tgz2.编辑配置文件[root@bigdatasoftware]#mvkafka_2.11-0.10.0.0kafka[root@bigdatasoftware]#cdkafka/config/[root@
首先说明，本人之前没用过zookeeper、kafka等，尚硅谷十几个小时的教程实在没有耐心看，现在我也不知道分区、副本之类的概念。用kafka只是听说他比RabbitMQ快，我也是昨天晚上刚使用，下文中若有讲错的地方或者我的理解与它的本质有偏差的地方请包涵。此文背景的环境是windows，linux流程也差不多。官网下载kafka，选择BinarydownloadsApacheKafka解压在D盘下或者什么地方，注意不要放在桌面等绝对路径太长的地方打开config中的zookeeper.properties，自己选择性修改clientPort，不想改也行修改config中的server.pr
启动zookeeper、kafka并创建kafka主题./bin/zkServer.shstart./bin/kafka-server-start.sh-daemon./config/server.properties./bin/kafka-topic.sh--create--topichunter--partitions3--replication-factor1--zookeeperlocalhost:90922、创建flume-kafka.conf配置文件用于采集socket数据后存入kafka在flume文件夹中的conf下新建flume-kafka.conf配置文件vimflume-
前言跟着尚硅谷海哥文档搭建的Kafka集群环境，在此记录一下，侵删注意：博主在服务器上搭建环境的时候使用的是一个服务器，所以这篇博客可能会出现一些xsync分发到其他服务器时候的错误，如果你在搭建的过程中出现了错误，欢迎评论来访，我们一起解决。准备工作准备三台服务器：hadoop102，hadoop103，hadoop104，在opt文件下先创建两个文件module和softwareHadoop部分(Hadoop如果不使用的话，可以不用安装Hadoop，但是在此阶段的环境搭建还要进行)JDK的安装1、用XShell传输工具将JDK导入到opt目录下的software文件夹下面2、在softwa
一清空kafka数据1关闭kafka2删除kafka存储目录（1）查询server.properties文件log.dirs配置，默认为“/tmp/kafka-logs”（2）删除目录3删除zookeeper上与kafka相关的znode节点（1）连接zookeeper./zkCli.sh-server127.0.0.1:2181//备注：默认端口是2128。具体可在kafkaserver.properties文件查询zookeeper.connect字段（2）删除consumers目录ls/rmr/consumers4重启zookeeper和kafka
数据传递语义至少一次：ACK级别设置为-1+分区副本大于等于2+ISR里应答的最小副本数量大于等于2最多一次：ACK级别设置为0总结：AtLeastOnce：可以保证数据不丢失，但是不能保证数据不重复AtMostOnce：可以保证数据不重复，但是不能保证数据不丢失精确一次：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失，Kafka0.11版本以后，引入了重大特性：幂等性和事务幂等性幂等性就是值Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证不重复。精确一次=幂等性+至少一次（ack=-1+分区副本数>=2+ISR最小副本数量>=
一、在java中配置pomjunitjunit4.11testorg.apache.kafkakafka-clients2.8.0org.apache.kafkakafka_2.122.8.0二、生产者方法(1)、ProducerJava中写在生产者输入内容在kafka中可以让消费者提取[root@kb144config]#kafka-console-consumer.sh--bootstrap-server192.168.153.144:9092--topickb22packagenj.zb.kb22.Kafka;importorg.apache.kafka.clients.producer
kafka3.4.0版本升级–helm部署前言最近由于kafka的漏洞需要升级至3.4.0版本，之前由于不是helm部署，升级起来出现了权限问题、挂盘问题，在k8s搞了许久都搞不定，狠下心来，直接来一波helm安装，在2月份的时候，helm官网已推出chart-21.0.1包（https://artifacthub.io/packages/helm/bitnami/kafka）。用chart-21.0.1包准备开搞。helm安装kafkahelm安装kafka比较简单，便不多说，官网有相应的说明，简单来说就是下载个chart包，在helm环境下执行helm安装即可。helmrepoaddbit
我的前一篇博客《kafka:AdminClient获取指定主题的所有消费者的消费偏移(一)》为了忽略忽略掉上线之前的所有消息，从获取指定主题的所有消费者的消费偏移并计算出最大偏移来解决此问题。但这个方案需要使用不常用的AdminClient类，而且如果该主题如果是第一次被消费者拉取消息时，因为得不到消费者的消费偏移，最后的结果，就是从0偏移开始拉取所有消息。并不能真正实现忽略上线之前所有消息的目的。所以我又优化了方案。基本的原理就是使用KafkaConsumer.offsetsForTimes方法获取消费者的所有主题分区的指定时间的偏移，并将这个偏移作为消费开始的偏移(KafkaConsume
