Kafka安装与使用

今天药忘吃喽~ 2022-04-03 05:27 333阅读 0赞

2.1. 下载

你可以在kafka官网 http://kafka.apache.org/downloads下载到最新的kafka安装包，选择下载二进制版本的tgz文件，根据网络状态可能需要fq，这里我们选择的版本是kafka\_2.11-1.1.0，目前的最新版

2.2. 安装

Kafka是使用scala编写的运行与jvm虚拟机上的程序，虽然也可以在windows上使用，但是kafka基本上是运行在linux服务器上，因此我们这里也使用linux来开始今天的实战。

首先确保你的机器上安装了jdk，kafka需要java运行环境，以前的kafka还需要zookeeper，新版的kafka已经内置了一个zookeeper环境，所以我们可以直接使用

说是安装，如果只需要进行最简单的尝试的话我们只需要解压到任意目录即可，这里我们将kafka压缩包解压到/home目录

2.3. 配置

在kafka解压目录下下有一个config的文件夹，里面放置的是我们的配置文件

consumer.properites 消费者配置，这个配置文件用于配置于2.5节中开启的消费者，此处我们使用默认的即可

producer.properties 生产者配置，这个配置文件用于配置于2.5节中开启的生产者，此处我们使用默认的即可

server.properties kafka服务器的配置，此配置文件用来配置kafka服务器，目前仅介绍几个最基础的配置

broker.id 申明当前kafka服务器在集群中的唯一ID，需配置为integer,并且集群中的每一个kafka服务器的id都应是唯一的，我们这里采用默认配置即可  
listeners 申明此kafka服务器需要监听的端口号，如果是在本机上跑虚拟机运行可以不用配置本项，默认会使用localhost的地址，如果是在远程服务器上运行则必须配置，例如：listeners=PLAINTEXT://192.168.180.128:9092。并确保服务器的9092端口能够访问  
zookeeper.connect 申明kafka所连接的zookeeper的地址 ，需配置为zookeeper的地址，由于本次使用的是kafka高版本中自带zookeeper，使用默认配置即可  
zookeeper.connect=localhost:2181  
当我们有多个应用,在不同的应用中都使用zookeer,都使用默认的zk端口的话就会2181端口冲突,我们可以设置自己的端口号,在config文件夹下zookeeper.properties文件中修改为

clientPort=2185

也就是zk开放接口为2185.

同时修改kafka的接入端口,server.properties文件中修改为

zookeeper.connect=localhost:2185

这样我们就成功修改了kafka里面的端口号

2.4. 运行

启动zookeeper

cd进入kafka解压目录，输入

bin/zookeeper-server-start.sh config/zookeeper.properties  
1  
启动zookeeper成功后会看到如下的输出

2.启动kafka

cd进入kafka解压目录，输入

bin/kafka-server-start.sh config/server.properties  
1  
启动kafka成功后会看到如下的输出

2.5. 第一个消息

2.5.1 创建一个topic

Kafka通过topic对同一类的数据进行管理，同一类的数据使用同一个topic可以在处理数据时更加的便捷

在kafka解压目录打开终端，输入

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test  
1  
创建一个名为test的topic

在创建topic后可以通过输入

bin/kafka-topics.sh --list --zookeeper localhost:2181  
1  
来查看已经创建的topic

2.5.2 创建一个消息消费者

在kafka解压目录打开终端，输入

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning  
1  
可以创建一个用于消费topic为test的消费者

消费者创建完成之后，因为还没有发送任何数据，因此这里在执行后没有打印出任何数据

不过别着急，不要关闭这个终端，打开一个新的终端，接下来我们创建第一个消息生产者

2.5.3 创建一个消息生产者

在kafka解压目录打开一个新的终端，输入

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test  
1  
在执行完毕后会进入的编辑器页面

在发送完消息之后，可以回到我们的消息消费者终端中，可以看到，终端中已经打印出了我们刚才发送的消息

kafka清理数据和topic

1、删除kafka存储目录（server.properties文件log.dirs配置，默认为"/tmp/kafka-logs"）相关topic目录

2、Kafka 删除topic的命令是：

./bin/kafka-topics  --delete --zookeeper 【zookeeper server】  --topic 【topic name】  
1  
如果kafaka启动时加载的配置文件中server.properties没有配置delete.topic.enable=true，那么此时的删除并不是真正的删除，而是把topic标记为：marked for deletion

你可以通过命令：

./bin/kafka-topics --zookeeper 【zookeeper server】 --list 来查看所有topic  
1  
此时你若想真正删除它，可以如下操作：

（1）登录zookeeper客户端：命令：./bin/zookeeper-client

（2）找到topic所在的目录：ls /brokers/topics

（3）找到要删除的topic，执行命令：rmr /brokers/topics/【topic name】即可，此时topic被彻底删除。

另外被标记为marked for deletion的topic你可以在zookeeper客户端中通过命令获得：ls /admin/delete\_topics/【topic name】，

如果你删除了此处的topic，那么marked for deletion 标记消失

zookeeper 的config中也有有关topic的信息： ls /config/topics/【topic name】暂时不知道有什么用

总结：

彻底删除topic：

1、删除kafka存储目录（server.properties文件log.dirs配置，默认为"/tmp/kafka-logs"）相关topic目录

2、如果配置了delete.topic.enable=true直接通过命令删除，如果命令删除不掉，直接通过zookeeper-client 删除掉broker下的topic即可。

python操作kafka

我们已经知道了kafka是一个消息队列,下面我们来学习怎么向kafka中传递数据和如何从kafka中获取数据

首先安装python的kafka库

pip install kafka  
1  
按照官网的样例，先跑一个应用

1、生产者:

from kafka import KafkaProducer  
import time  
producer = KafkaProducer(bootstrap\_servers=\['127.0.0.1:9092'\])  \#此处ip可以是多个\['0.0.0.1:9092','0.0.0.2:9092','0.0.0.3:9092' \]

i=0  
while True:  
    i+=1  
    msg = "producer1+%d" % i  
    print(msg)  
    producer.send('test', msg.encode('utf-8'))  \# 参数为主题和bytes数据  
    time.sleep(1)

producer.close()  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
启动后生产者便可以将字节流发送到kafka服务器.

2、消费者(简单demo):

from kafka import KafkaConsumer

consumer = KafkaConsumer('test',bootstrap\_servers=\['127.0.0.1:9092'\])  \#参数为接收主题和kafka服务器地址

\# 这是一个永久堵塞的过程，生产者消息会缓存在消息队列中,并且不删除,所以每个消息在消息队列中都有偏移  
for message in consumer:  \# consumer是一个消息队列，当后台有消息时，这个消息队列就会自动增加．所以遍历也总是会有数据，当消息队列中没有数据时，就会堵塞等待消息带来  
    print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))

1  
2  
3  
4  
5  
6  
7  
8  
9  
启动后消费者可以从kafka服务器获取数据.

3、消费者(消费群组)

from kafka import KafkaConsumer  
\# 使用group,对于同一个group的成员只有一个消费者实例可以读取数据  
consumer = KafkaConsumer('test',group\_id='my-group',bootstrap\_servers=\['127.0.0.1:9092'\])  
for message in consumer:  
    print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))

1  
2  
3  
4  
5  
6  
启动多个消费者，只有其中某一个成员可以消费到，满足要求，消费组可以横向扩展提高处理能力

4、消费者(读取目前最早可读的消息)

from kafka import KafkaConsumer

consumer = KafkaConsumer('test',auto\_offset\_reset='earliest',bootstrap\_servers=\['127.0.0.1:9092'\])

for message in consumer:  
    print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))

1  
2  
3  
4  
5  
6  
7  
auto\_offset\_reset:重置偏移量，earliest移到最早的可用消息，latest最新的消息，默认为latest  
源码定义:\{‘smallest’: ‘earliest’, ‘largest’: ‘latest’\}

5、消费者(手动设置偏移量)

\# ==========读取指定位置消息===============  
from kafka import KafkaConsumer  
from kafka.structs import TopicPartition

consumer = KafkaConsumer('test',bootstrap\_servers=\['127.0.0.1:9092'\])

print(consumer.partitions\_for\_topic("test"))  \#获取test主题的分区信息  
print(consumer.topics())  \#获取主题列表  
print(consumer.subscription())  \#获取当前消费者订阅的主题  
print(consumer.assignment())  \#获取当前消费者topic、分区信息  
print(consumer.beginning\_offsets(consumer.assignment())) \#获取当前消费者可消费的偏移量  
consumer.seek(TopicPartition(topic='test', partition=0), 5)  \#重置偏移量，从第5个偏移量消费  
for message in consumer:  
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
6、消费者(订阅多个主题)

\# =======订阅多个消费者==========

from kafka import KafkaConsumer  
from kafka.structs import TopicPartition

consumer = KafkaConsumer(bootstrap\_servers=\['127.0.0.1:9092'\])  
consumer.subscribe(topics=('test','test0'))  \#订阅要消费的主题  
print(consumer.topics())  
print(consumer.position(TopicPartition(topic='test', partition=0))) \#获取当前主题的最新偏移量  
for message in consumer:  
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
7、消费者(手动拉取消息)

from kafka import KafkaConsumer  
import time

consumer = KafkaConsumer(bootstrap\_servers=\['127.0.0.1:9092'\])  
consumer.subscribe(topics=('test','test0'))  
while True:  
    msg = consumer.poll(timeout\_ms=5)   \#从kafka获取消息  
    print(msg)  
    time.sleep(2)

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
8、消费者(消息挂起与恢复)

\# ==============消息恢复和挂起===========

from kafka import KafkaConsumer  
from kafka.structs import TopicPartition  
import time

consumer = KafkaConsumer(bootstrap\_servers=\['127.0.0.1:9092'\])  
consumer.subscribe(topics=('test'))  
consumer.topics()  
consumer.pause(TopicPartition(topic=u'test', partition=0))  \# pause执行后，consumer不能读取，直到调用resume后恢复。  
num = 0  
while True:  
    print(num)  
    print(consumer.paused())   \#获取当前挂起的消费者  
    msg = consumer.poll(timeout\_ms=5)  
    print(msg)  
    time.sleep(2)  
    num = num + 1  
    if num == 10:  
        print("resume...")  
        consumer.resume(TopicPartition(topic='test', partition=0))  
        print("resume......")  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
pause执行后，consumer不能读取，直到调用resume后恢复。

Python创建自定义的Kafka Topic

client = KafkaClient(bootstrap\_servers=brokers)

if topic not in client.cluster.topics(exclude\_internal\_topics=True):  \# Topic不存在

request = admin.CreateTopicsRequest\_v0(  
        create\_topic\_requests=\[(  
            topic,  
            num\_partitions,  
            -1,  \# replication unset.  
            \[\],  \# Partition assignment.  
            \[(key, value) for key, value in configs.items()\],  \# Configs  
        )\],  
        timeout=timeout\_ms  
    )

future = client.send(2, request)  \# 2是Controller,发送给其他Node都创建失败。  
    client.poll(timeout\_ms=timeout\_ms, future=future, sleep=False)  \# 这里

result = future.value  
    \# error\_code = result.topic\_error\_codes\[0\]\[1\]  
    print("CREATE TOPIC RESPONSE: ", result)  \# 0 success, 41 NOT\_CONTROLLER, 36 ALREADY\_EXISTS  
    client.close()  
else:  \# Topic已经存在  
    print("Topic already exists!")  
    return

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
kafka的配置

在kafka/config/目录下面有3个配置文件：

producer.properties

consumer.properties

server.properties

kafka的配置分为 broker（server.properties）、producter（producer.properties）、consumer（consumer.properties）三个不同的配置

一 BROKER 的全局配置

最为核心的三个配置 broker.id、log.dir、zookeeper.connect 。

\------------------------------------------- 系统 相关 -------------------------------------------  
\#\#每一个broker在集群中的唯一标示，要求是正数。在改变IP地址，不改变broker.id的话不会影响consumers  
broker.id =1  
   
\#\#kafka数据的存放地址，多个地址的话用逗号分割 /tmp/kafka-logs-1，/tmp/kafka-logs-2  
log.dirs = /tmp/kafka-logs  
   
\#\#提供给客户端响应的端口  
port =6667  
   
\#\#消息体的最大大小，单位是字节  
message.max.bytes =1000000  
   
\#\# broker 处理消息的最大线程数，一般情况下不需要去修改  
num.network.threads =3  
   
\#\# broker处理磁盘IO 的线程数 ，数值应该大于你的硬盘数  
num.io.threads =8  
   
\#\# 一些后台任务处理的线程数，例如过期消息文件的删除等，一般情况下不需要去做修改  
background.threads =4  
   
\#\# 等待IO线程处理的请求队列最大数，若是等待IO的请求超过这个数值，那么会停止接受外部消息，算是一种自我保护机制  
queued.max.requests =500  
   
\#\#broker的主机地址，若是设置了，那么会绑定到这个地址上，若是没有，会绑定到所有的接口上，并将其中之一发送到ZK，一般不设置  
host.name  
   
\#\# 打广告的地址，若是设置的话，会提供给producers, consumers,其他broker连接，具体如何使用还未深究  
advertised.host.name  
   
\#\# 广告地址端口，必须不同于port中的设置  
advertised.port  
   
\#\# socket的发送缓冲区，socket的调优参数SO\_SNDBUFF  
socket.send.buffer.bytes =100\*1024  
   
\#\# socket的接受缓冲区，socket的调优参数SO\_RCVBUFF  
socket.receive.buffer.bytes =100\*1024  
   
\#\# socket请求的最大数值，防止serverOOM，message.max.bytes必然要小于socket.request.max.bytes，会被topic创建时的指定参数覆盖  
socket.request.max.bytes =100\*1024\*1024  
   
\------------------------------------------- LOG 相关 -------------------------------------------  
\#\# topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，会被topic创建时的指定参数覆盖  
log.segment.bytes =1024\*1024\*1024  
   
\#\# 这个参数会在日志segment没有达到log.segment.bytes设置的大小，也会强制新建一个segment 会被 topic创建时的指定参数覆盖  
log.roll.hours =24\*7  
   
\#\# 日志清理策略 选择有：delete和compact 主要针对过期数据的处理，或是日志文件达到限制的额度，会被 topic创建时的指定参数覆盖  
log.cleanup.policy = delete  
   
\#\# 数据存储的最大时间 超过这个时间 会根据log.cleanup.policy设置的策略处理数据，也就是消费端能够多久去消费数据  
\#\# log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖  
log.retention.minutes=7days

指定日志每隔多久检查看是否可以被删除，默认1分钟  
log.cleanup.interval.mins=1  
   
\#\# topic每个分区的最大文件大小，一个topic的大小限制 = 分区数\*log.retention.bytes 。-1没有大小限制  
\#\# log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖  
log.retention.bytes=-1  
   
\#\# 文件大小检查的周期时间，是否处罚 log.cleanup.policy中设置的策略  
log.retention.check.interval.ms=5minutes  
   
\#\# 是否开启日志压缩  
log.cleaner.enable=false  
   
\#\# 日志压缩运行的线程数  
log.cleaner.threads =1  
   
\#\# 日志压缩时候处理的最大大小  
log.cleaner.io.max.bytes.per.second=None  
   
\#\# 日志压缩去重时候的缓存空间 ，在空间允许的情况下，越大越好  
log.cleaner.dedupe.buffer.size=500\*1024\*1024  
   
\#\# 日志清理时候用到的IO块大小 一般不需要修改  
log.cleaner.io.buffer.size=512\*1024  
   
\#\# 日志清理中hash表的扩大因子 一般不需要修改  
log.cleaner.io.buffer.load.factor =0.9  
   
\#\# 检查是否处罚日志清理的间隔  
log.cleaner.backoff.ms =15000  
   
\#\# 日志清理的频率控制，越大意味着更高效的清理，同时会存在一些空间上的浪费，会被topic创建时的指定参数覆盖  
log.cleaner.min.cleanable.ratio=0.5  
   
\#\# 对于压缩的日志保留的最长时间，也是客户端消费消息的最长时间，同log.retention.minutes的区别在于一个控制未压缩数据，一个控制压缩后的数据。会被topic创建时的指定参数覆盖  
log.cleaner.delete.retention.ms =1day  
   
\#\# 对于segment日志的索引文件大小限制，会被topic创建时的指定参数覆盖  
log.index.size.max.bytes =10\*1024\*1024  
   
\#\# 当执行一个fetch操作后，需要一定的空间来扫描最近的offset大小，设置越大，代表扫描速度越快，但是也更好内存，一般情况下不需要搭理这个参数  
log.index.interval.bytes =4096  
   
\#\# log文件"sync"到磁盘之前累积的消息条数  
\#\# 因为磁盘IO操作是一个慢操作,但又是一个"数据可靠性"的必要手段  
\#\# 所以此参数的设置,需要在"数据可靠性"与"性能"之间做必要的权衡.  
\#\# 如果此值过大,将会导致每次"fsync"的时间较长(IO阻塞)  
\#\# 如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的client请求有一定的延迟.  
\#\# 物理server故障,将会导致没有fsync的消息丢失.  
log.flush.interval.messages=None  
   
\#\# 检查是否需要固化到硬盘的时间间隔  
log.flush.scheduler.interval.ms =3000  
   
\#\# 仅仅通过interval来控制消息的磁盘写入时机,是不足的.  
\#\# 此参数用于控制"fsync"的时间间隔,如果消息量始终没有达到阀值,但是离上一次磁盘同步的时间间隔  
\#\# 达到阀值,也将触发.  
log.flush.interval.ms = None  
   
\#\# 文件在索引中清除后保留的时间 一般不需要去修改  
log.delete.delay.ms =60000  
   
\#\# 控制上次固化硬盘的时间点，以便于数据恢复 一般不需要去修改  
log.flush.offset.checkpoint.interval.ms =60000  
   
\------------------------------------------- TOPIC 相关 -------------------------------------------  
\#\# 是否允许自动创建topic ，若是false，就需要通过命令创建topic  
auto.create.topics.enable =true  
   
\#\# 一个topic ，默认分区的replication个数 ，不得大于集群中broker的个数  
default.replication.factor =1  
   
\#\# 每个topic的分区个数，若是在topic创建时候没有指定的话 会被topic创建时的指定参数覆盖  
num.partitions =1  
   
实例 --replication-factor3--partitions1--topic replicated-topic ：名称replicated-topic有一个分区，分区被复制到三个broker上。  
   
\----------------------------------复制(Leader、replicas) 相关 ----------------------------------  
\#\# partition leader与replicas之间通讯时,socket的超时时间  
controller.socket.timeout.ms =30000  
   
\#\# partition leader与replicas数据同步时,消息的队列尺寸  
controller.message.queue.size=10  
   
\#\# replicas响应partition leader的最长等待时间，若是超过这个时间，就将replicas列入ISR(in-sync replicas)，并认为它是死的，不会再加入管理中  
replica.lag.time.max.ms =10000  
   
\#\# 如果follower落后与leader太多,将会认为此follower\[或者说partition relicas\]已经失效  
\#\# 通常,在follower与leader通讯时,因为网络延迟或者链接断开,总会导致replicas中消息同步滞后  
\#\# 如果消息之后太多,leader将认为此follower网络延迟较大或者消息吞吐能力有限,将会把此replicas迁移  
\#\# 到其他follower中.  
\#\# 在broker数量较少,或者网络不足的环境中,建议提高此值.  
replica.lag.max.messages =4000  
   
\#\#follower与leader之间的socket超时时间  
replica.socket.timeout.ms=30\*1000  
   
\#\# leader复制时候的socket缓存大小  
replica.socket.receive.buffer.bytes=64\*1024  
   
\#\# replicas每次获取数据的最大大小  
replica.fetch.max.bytes =1024\*1024  
   
\#\# replicas同leader之间通信的最大等待时间，失败了会重试  
replica.fetch.wait.max.ms =500  
   
\#\# fetch的最小数据尺寸,如果leader中尚未同步的数据不足此值,将会阻塞,直到满足条件  
replica.fetch.min.bytes =1  
   
\#\# leader 进行复制的线程数，增大这个数值会增加follower的IO  
num.replica.fetchers=1  
   
\#\# 每个replica检查是否将最高水位进行固化的频率  
replica.high.watermark.checkpoint.interval.ms =5000  
   
\#\# 是否允许控制器关闭broker ,若是设置为true,会关闭所有在这个broker上的leader，并转移到其他broker  
controlled.shutdown.enable =false  
   
\#\# 控制器关闭的尝试次数  
controlled.shutdown.max.retries =3  
   
\#\# 每次关闭尝试的时间间隔  
controlled.shutdown.retry.backoff.ms =5000  
   
\#\# 是否自动平衡broker之间的分配策略  
auto.leader.rebalance.enable =false  
   
\#\# leader的不平衡比例，若是超过这个数值，会对分区进行重新的平衡  
leader.imbalance.per.broker.percentage =10  
   
\#\# 检查leader是否不平衡的时间间隔  
leader.imbalance.check.interval.seconds =300  
   
\#\# 客户端保留offset信息的最大空间大小  
offset.metadata.max.bytes  
   
\----------------------------------ZooKeeper 相关----------------------------------  
\#\#zookeeper集群的地址，可以是多个，多个之间用逗号分割 hostname1:port1,hostname2:port2,hostname3:port3  
zookeeper.connect = localhost:2181  
   
\#\# ZooKeeper的最大超时时间，就是心跳的间隔，若是没有反映，那么认为已经死了，不易过大  
zookeeper.session.timeout.ms=6000  
   
\#\# ZooKeeper的连接超时时间  
zookeeper.connection.timeout.ms =6000  
   
\#\# ZooKeeper集群中leader和follower之间的同步实际那  
zookeeper.sync.time.ms =2000  
配置的修改  
其中一部分配置是可以被每个topic自身的配置所代替，例如  
新增配置  
bin/kafka-topics.sh --zookeeper localhost:2181--create --topic my-topic --partitions1--replication-factor1--config max.message.bytes=64000--config flush.messages=1  
   
修改配置  
bin/kafka-topics.sh --zookeeper localhost:2181--alter --topic my-topic --config max.message.bytes=128000  
   
删除配置 ：  
bin/kafka-topics.sh --zookeeper localhost:2181--alter --topic my-topic --deleteConfig max.message.bytes

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
33  
34  
35  
36  
37  
38  
39  
40  
41  
42  
43  
44  
45  
46  
47  
48  
49  
50  
51  
52  
53  
54  
55  
56  
57  
58  
59  
60  
61  
62  
63  
64  
65  
66  
67  
68  
69  
70  
71  
72  
73  
74  
75  
76  
77  
78  
79  
80  
81  
82  
83  
84  
85  
86  
87  
88  
89  
90  
91  
92  
93  
94  
95  
96  
97  
98  
99  
100  
101  
102  
103  
104  
105  
106  
107  
108  
109  
110  
111  
112  
113  
114  
115  
116  
117  
118  
119  
120  
121  
122  
123  
124  
125  
126  
127  
128  
129  
130  
131  
132  
133  
134  
135  
136  
137  
138  
139  
140  
141  
142  
143  
144  
145  
146  
147  
148  
149  
150  
151  
152  
153  
154  
155  
156  
157  
158  
159  
160  
161  
162  
163  
164  
165  
166  
167  
168  
169  
170  
171  
172  
173  
174  
175  
176  
177  
178  
179  
180  
181  
182  
183  
184  
185  
186  
187  
188  
189  
190  
191  
192  
193  
194  
195  
196  
197  
198  
199  
200  
201  
202  
203  
204  
205  
206  
207  
208  
209  
210  
211  
212  
213  
214  
215  
216  
217  
218  
219  
二 CONSUMER 配置

最为核心的配置是group.id、zookeeper.connect

\#\# Consumer归属的组ID，broker是根据group.id来判断是队列模式还是发布订阅模式，非常重要  
 group.id  
   
\#\# 消费者的ID，若是没有设置的话，会自增  
 consumer.id  
   
\#\# 一个用于跟踪调查的ID ，最好同group.id相同  
 client.id = group id value  
   
\#\# 对于zookeeper集群的指定，可以是多个 hostname1:port1,hostname2:port2,hostname3:port3 必须和broker使用同样的zk配置  
 zookeeper.connect=localhost:2182  
   
\#\# zookeeper的心跳超时时间，查过这个时间就认为是dead消费者  
 zookeeper.session.timeout.ms =6000  
   
\#\# zookeeper的等待连接时间  
 zookeeper.connection.timeout.ms =6000  
   
\#\# zookeeper的follower同leader的同步时间  
 zookeeper.sync.time.ms =2000  
   
\#\# 当zookeeper中没有初始的offset时候的处理方式 。smallest ：重置为最小值 largest:重置为最大值 anythingelse：抛出异常  
 auto.offset.reset = largest  
   
\#\# socket的超时时间，实际的超时时间是：max.fetch.wait + socket.timeout.ms.  
 socket.timeout.ms=30\*1000  
   
\#\# socket的接受缓存空间大小  
 socket.receive.buffer.bytes=64\*1024  
   
\#\#从每个分区获取的消息大小限制  
 fetch.message.max.bytes =1024\*1024  
   
\#\# 是否在消费消息后将offset同步到zookeeper，当Consumer失败后就能从zookeeper获取最新的offset  
 auto.commit.enable =true  
   
\#\# 自动提交的时间间隔  
 auto.commit.interval.ms =60\*1000  
   
\#\# 用来处理消费消息的块，每个块可以等同于fetch.message.max.bytes中数值  
 queued.max.message.chunks =10  
   
\#\# 当有新的consumer加入到group时,将会reblance,此后将会有partitions的消费端迁移到新  
\#\# 的consumer上,如果一个consumer获得了某个partition的消费权限,那么它将会向zk注册  
\#\#"Partition Owner registry"节点信息,但是有可能此时旧的consumer尚没有释放此节点,  
\#\# 此值用于控制,注册节点的重试次数.  
 rebalance.max.retries =4  
   
\#\# 每次再平衡的时间间隔  
 rebalance.backoff.ms =2000  
   
\#\# 每次重新选举leader的时间  
 refresh.leader.backoff.ms  
   
\#\# server发送到消费端的最小数据，若是不满足这个数值则会等待，知道满足数值要求  
 fetch.min.bytes =1  
   
\#\# 若是不满足最小大小(fetch.min.bytes)的话，等待消费端请求的最长等待时间  
 fetch.wait.max.ms =100  
   
\#\# 指定时间内没有消息到达就抛出异常，一般不需要改  
 consumer.timeout.ms = -1

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
33  
34  
35  
36  
37  
38  
39  
40  
41  
42  
43  
44  
45  
46  
47  
48  
49  
50  
51  
52  
53  
54  
55  
56  
57  
58  
59  
60  
61  
62  
63  
64  
三 PRODUCER 的配置

比较核心的配置：metadata.broker.list、request.required.acks、producer.type、serializer.class

\#\# 消费者获取消息元信息(topics, partitions and replicas)的地址,配置格式是：host1:port1,host2:port2，也可以在外面设置一个vip  
 metadata.broker.list  
   
\#\#消息的确认模式  
 \#\#0：不保证消息的到达确认，只管发送，低延迟但是会出现消息的丢失，在某个server失败的情况下，有点像TCP  
 \#\#1：发送消息，并会等待leader 收到确认后，一定的可靠性  
 \#\# -1：发送消息，等待leader收到确认，并进行复制操作后，才返回，最高的可靠性  
 request.required.acks =0  
   
\#\# 消息发送的最长等待时间  
 request.timeout.ms =10000  
   
\#\# socket的缓存大小  
 send.buffer.bytes=100\*1024  
   
\#\# key的序列化方式，若是没有设置，同serializer.class  
 key.serializer.class  
   
\#\# 分区的策略，默认是取模  
 partitioner.class=kafka.producer.DefaultPartitioner  
   
\#\# 消息的压缩模式，默认是none，可以有gzip和snappy  
 compression.codec = none  
   
\#\# 可以针对默写特定的topic进行压缩  
 compressed.topics=null  
   
\#\# 消息发送失败后的重试次数  
 message.send.max.retries =3  
   
\#\# 每次失败后的间隔时间  
 retry.backoff.ms =100  
   
\#\# 生产者定时更新topic元信息的时间间隔 ，若是设置为0，那么会在每个消息发送后都去更新数据  
 topic.metadata.refresh.interval.ms =600\*1000  
   
\#\# 用户随意指定，但是不能重复，主要用于跟踪记录消息  
 client.id=""  
   
\------------------------------------------- 消息模式 相关 -------------------------------------------  
 \#\# 生产者的类型 async:异步执行消息的发送 sync：同步执行消息的发送  
 producer.type=sync  
   
\#\# 异步模式下，那么就会在设置的时间缓存消息，并一次性发送  
 queue.buffering.max.ms =5000  
   
\#\# 异步的模式下 最长等待的消息数  
 queue.buffering.max.messages =10000  
   
\#\# 异步模式下，进入队列的等待时间 若是设置为0，那么要么进入队列，要么直接抛弃  
 queue.enqueue.timeout.ms = -1  
   
\#\# 异步模式下，每次发送的最大消息数，前提是触发了queue.buffering.max.messages或是queue.buffering.max.ms的限制  
 batch.num.messages=200  
   
\#\# 消息体的系列化处理类 ，转化为字节流进行传输  
 serializer.class= kafka.serializer.DefaultEncoder  
\---------------------   
作者：数据架构师   
来源：CSDN   
原文：https://blog.csdn.net/luanpeng825485697/article/details/81036028   
版权声明：本文为博主原创文章，转载请附上博文链接！