元数据信息

/opt/app/kafka/bin]$./kafka-topics.sh

--create

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--create

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--list

pxj62:2181,pxj63:2181,pxj64:2181

test01

/opt/app/kafka/bin]$./kafka-topics.sh

--describe

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--describe

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--create

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--describe

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--alter

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--describe

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--delete

pxj62:2181,pxj63:2181,pxj64:2181

--topic

/opt/app/kafka/bin]$./kafka-topics.sh

--describe

pxj62:2181,pxj63:2181,pxj64:2181

--topic

java.lang.IllegalArgumentException:

Topic

kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:484)at

kafka.admin.TopicCommand$ZookeeperTopicService.describeTopic(TopicCommand.scala:390)at

kafka.admin.TopicCommand$.main(TopicCommand.scala:67)at

kafka.admin.TopicCommand.main(TopicCommand.scala)(kafka.admin.TopicCommand$)

[pxjpxj62

/opt/app/kafka/bin]$./kafka-console-producer.sh

--broker-list

pxj62:9092,pxj63:9092,pxj64:9092

--topic

/opt/app/kafka/bin]$./kafka-console-consumer.sh

--bootstrap-server

pxj62:9092,pxj63:9092,pxj64:9092

--topic

org.apache.kafka.clients.producer.KafkaProducer;

import

org.apache.kafka.clients.producer.Producer;

import

org.apache.kafka.clients.producer.ProducerRecord;import

class

Properties();props.put(bootstrap.servpackage

com.ccj.pxj.kafka;

org.apache.kafka.clients.consumer.ConsumerRecord;

import

org.apache.kafka.clients.consumer.ConsumerRecords;

import

org.apache.kafka.clients.consumer.KafkaConsumer;import

java.time.Duration;

Properties();props.setProperty(bootstrap.servers,

pxj62:9092,pxj63:9092,pxj64:9092);

kafka地址props.setProperty(group.id,

test);

idprops.setProperty(enable.auto.commit,

true);

是否开启自动提交数据的偏移量props.setProperty(auto.commit.interval.ms,

1000);

自动提交的间隔时间props.setProperty(key.deserializer,

org.apache.kafka.common.serialization.StringDeserializer);

设置key反序列类props.setProperty(value.deserializer,

org.apache.kafka.common.serialization.StringDeserializer);//

设置value反序列化类//1.2:

创建kafka消费者对象KafkaConsumerString,

String

KafkaConsumer(props);//2.设置消费者监听那些Topicconsumer.subscribe(Arrays.asList(test02));//3.

消费数据:

参数表示等待(超时)的时间ConsumerRecordsString,

String

consumer.poll(Duration.ofMillis(100));for

String

从哪个分区读取的数据System.out.println(偏移量:

offset

pxj62:9092,pxj63:9092,pxj64:9092);

all);

指定消息确认方案props.put(key.serializer,

org.apache.kafka.common.serialization.StringSerializer);//

key序列化类props.put(value.serializer,

org.apache.kafka.common.serialization.StringSerializer);

value序列化类//1.2:

ProducerRecord(test02,Integer.toString(i));producer.send(producerRecord);}//3.

}5.2

org.apache.kafka.clients.consumer.ConsumerRecord;

import

org.apache.kafka.clients.consumer.ConsumerRecords;

import

org.apache.kafka.clients.consumer.KafkaConsumer;import

java.time.Duration;

Properties();props.setProperty(bootstrap.servers,

pxj62:9092,pxj63:9092,pxj64:9092);

kafka地址props.setProperty(group.id,

test);

idprops.setProperty(enable.auto.commit,

true);

是否开启自动提交数据的偏移量props.setProperty(auto.commit.interval.ms,

1000);

自动提交的间隔时间props.setProperty(key.deserializer,

org.apache.kafka.common.serialization.StringDeserializer);

设置key反序列类props.setProperty(value.deserializer,

org.apache.kafka.common.serialization.StringDeserializer);//

设置value反序列化类//1.2:

创建kafka消费者对象KafkaConsumerString,

String

KafkaConsumer(props);//2.设置消费者监听那些Topicconsumer.subscribe(Arrays.asList(test02));//3.

消费数据:

参数表示等待(超时)的时间ConsumerRecordsString,

String

consumer.poll(Duration.ofMillis(100));for

String

从哪个分区读取的数据System.out.println(偏移量:

offset

org.slf4j.impl.StaticLoggerBinder.

SLF4J:

http://www.slf4j.org/codes.html#StaticLoggerBinder

for

1生产者只管发送到broket端需要等待对应接受分片的主副本接收到数据后给予响应认为数据发送成功

-1:ALL生产者只管发送到broket端需要等待对应接受分片所有的

副本接收到数据后给予响应认为数据发送成功

props.put(“acks”,all)6.3如果broker端迟迟没有给予响应如何解决

采用先等待超时时间再重试的策略一般重试3次如果重试后依然没有给予响应此时让程序直接报错。

通知相关人员处理即可

6.4宽带占用如何解决

可以引入缓存池采用异步发送方案生产者将数据在发送数据时候底层会将这个数据保存到缓存池中当池子中数据达到一批数据大小后将达一批数据直接发送到broker此时broker针对这一批数据给予一次性响应即可批量发送数据6.5

采用批量发送数据如果发送一批数据到broker端broker端又没有给予响应此时缓存池中数据满了如何解决呢

解决方案

1.丢弃缓存池中数据报异常适用于数据不重要或者可以重读的消息总数据

2.在写入缓冲池的时候需要将数据在其他的地方也持久存储一份发送成功一批数据将持久化地方数据删除一部分以保证在出现此问题后数据依然存在下次启动的时候优先从持久化容器中读取即可七、安装

kafka-eagle

JAVA_HOME/usr/java/jdk1.8.0_141

export

ZOOKEEPER_HOME/opt/app/zookeeper

export

PATH${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${ZOOKEEPER_HOME}/bin:${KAFKA_HOME}/bin:${KE_HOME}/bin:$PATH[pxjpxj62

/home/pxj]$source

使用vi打开conf目录下的system-config.propertie

[pxjpxj62

kafka.eagle.zk.cluster.aliascluster1

cluster1.zk.listpxj62:2181,pxj63:2181,pxj64:2181

#cluster2.zk.listxdn10:2181,xdn11:2181,xdn12:2181

kafka

######################################

kafka

######################################

#kafka.eagle.driverorg.sqlite.JDBC

#kafka.eagle.urljdbc:sqlite:/hadoop/kafka-eagle/db/ke.db

#kafka.eagle.passwordwww.kafka-eagle.org######################################

kafka

######################################

kafka.eagle.drivercom.mysql.jdbc.Driver

kafka.eagle.urljdbc:mysql://pxj63:3306/ke?useUnicodetruecharacterEncodingUTF-8zeroDateTimeBehaviorconvertToNull

LOVE

JAVA_HOME/usr/java/jdk1.8.0_1417.5授权运行

[pxjpxj62

/opt/app/kafka-eagle/bin]$chmod

ke.sh

/opt/app/kafka-eagle/bin]$./ke.sh

start7.7访问web

org.apache.kafka.clients.producer.KafkaProducer;

import

org.apache.kafka.clients.producer.ProducerRecord;import

class

Properties();props.put(bootstrap.servers,

pxj62:9092,pxj63:9092,pxj64:9092);

all);

指定消息确认方案props.put(key.serializer,

org.apache.kafka.common.serialization.StringSerializer);//

key序列化类props.put(value.serializer,

org.apache.kafka.common.serialization.StringSerializer);

value序列化类//构造生产者KafkaProducerString,String

producer

get支持自定义超时的时间try{producer.send(producerRecord).get();}catch

(Exception

e){e.printStackTrace();}}producer.close();}

九、异步发送

org.apache.kafka.clients.producer.*;import

class

Properties();props.put(bootstrap.servers,

pxj62:9092,pxj63:9092,pxj64:9092);

all);

指定消息确认方案props.put(key.serializer,

org.apache.kafka.common.serialization.StringSerializer);//

key序列化类props.put(value.serializer,

org.apache.kafka.common.serialization.StringSerializer);

value序列化类//1.2:

i_22);producer.send(producerRecord,

new

编写发送失败的处理业务逻辑代码System.err.println(发送消息失败

e.getStackTrace());}if(metadata!null){if

(metadata

org.apache.kafka.clients.consumer.ConsumerRecord;

import

org.apache.kafka.clients.consumer.ConsumerRecords;

import

org.apache.kafka.clients.consumer.KafkaConsumer;import

java.time.Duration;

设置消费者的配置信息props.setProperty(bootstrap.servers,

pxj62:9092,pxj63:9092,pxj64:9092);

kafka地址props.setProperty(group.id,

test);

idprops.setProperty(enable.auto.commit,

false);

是否开启自动提交数据的偏移量props.setProperty(key.deserializer,

org.apache.kafka.common.serialization.StringDeserializer);

设置key反序列类props.setProperty(value.deserializer,

org.apache.kafka.common.serialization.StringDeserializer);//

设置value反序列化类

KafkaConsumer(props);consumer.subscribe(Arrays.asList(test01));while(true){ConsumerRecordsString,String

recordsconsumer.poll(Duration.ofMillis(100));for

String

从哪个分区读取的数据System.out.println(偏移量:

offset

会造成大量的重复消费问题consumer.commitSync();

异步提交}}}

__consumer_offset此topic具有50个分区,

默认值为

log.segment.bytes1073741824十二、kafka的数据查询机制

先确定这条消息在那个segment片段中到对应片段中找index文件,

根据offset查询消息数据在log文件的那个物理偏移量位置根据从index查询到的偏移量信息,

顺序读写

顺序读写效率远远高于随机读写十三、kafka中生产者的数据分发策略

kafka生产者数据分发策略:

stickyPartitionCache.partition(topic,

cluster);}

cluster.partitionsForTopic(topic);int

numPartitions

Utils.toPositive(Utils.murmur2(keyBytes))

numPartitions;}2)

stickyPartitionCache.partition(topic,

cluster);}

然后根据key进行hash取模ListPartitionInfo

partitions

cluster.partitionsForTopic(topic);int

numPartitions

Utils.toPositive(Utils.murmur2(keyBytes))

numPartitions;}3)

这种分区方案称为粘性分区策略粘性分区是kafka2.4.x及以上版本支持的一种全新的分区策略

将这一批数据写入到某一个topic中,

分别对应不同的分片,将各个小批次的数据发送给对应的分片下即可,

导致生产效率下降,

org.apache.kafka.clients.producer.internals.DefaultPartitioner通过生产者的properties对象,

当生产者开始发送数据,

这种分区方案称为粘性分区策略粘性分区是kafka2.4.x及以上版本支持的一种全新的分区策略

将这一批数据写入到某一个topic中,

分别对应不同的分片,将各个小批次的数据发送给对应的分片下即可,

导致生产效率下降,

一个消费只能被一个消费者所接收让所有监听这个topic的消费者都属于同一个消费者组内即可发布订阅:

一个消息可以被多个消费者所接收让所有监听这个topic的消费者都属于不同的消费者组内即可作者潘陈pxj

日期2023-04-30