本文主要是介绍Kafka broker配置介绍，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
这部分内容对了解系统和提高软件性能都有很大的帮助，kafka官网上也给出了比较详细的配置详单，但是我们还是直接从代码来看broker到底有哪些配置需要我们去了解的，配置都有英文注释，所以每一部分是干什么的就不翻译了，都能看懂：
| 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 | /** * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package kafka.server import java.util.Properties import kafka.utils.{Utils, ZKConfig} import kafka.message.Message /** * Configuration settings for the kafka server */ class KafkaConfig(props : Properties) extends ZKConfig(props) { /* the port to listen and accept connections on */ val port : Int = Utils.getInt(props, "port" , 6667 ) /* hostname of broker. If not set, will pick up from the value returned from getLocalHost. If there are multiple interfaces getLocalHost may not be what you want. */ val hostName : String = Utils.getString(props, "hostname" , null ) /* the broker id for this server */ val brokerId : Int = Utils.getInt(props, "brokerid" ) /* the SO_SNDBUFF buffer of the socket sever sockets */ val socketSendBuffer : Int = Utils.getInt(props, "socket.send.buffer" , 100 * 1024 ) /* the SO_RCVBUFF buffer of the socket sever sockets */ val socketReceiveBuffer : Int = Utils.getInt(props, "socket.receive.buffer" , 100 * 1024 ) /* the maximum number of bytes in a socket request */ val maxSocketRequestSize : Int = Utils.getIntInRange(props, "max.socket.request.bytes" , 100 * 1024 * 1024 , ( 1 , Int.MaxValue)) /* the maximum size of message that the server can receive */ val maxMessageSize = Utils.getIntInRange(props, "max.message.size" , 1000000 , ( 0 , Int.MaxValue)) /* the number of worker threads that the server uses for handling all client requests*/ val numThreads = Utils.getIntInRange(props, "num.threads" , Runtime.getRuntime().availableProcessors, ( 1 , Int.MaxValue)) /* the interval in which to measure performance statistics */ val monitoringPeriodSecs = Utils.getIntInRange(props, "monitoring.period.secs" , 600 , ( 1 , Int.MaxValue)) /* the default number of log partitions per topic */ val numPartitions = Utils.getIntInRange(props, "num.partitions" , 1 , ( 1 , Int.MaxValue)) /* the directory in which the log data is kept */ val logDir = Utils.getString(props, "log.dir" ) /* the maximum size of a single log file */ val logFileSize = Utils.getIntInRange(props, "log.file.size" , 1 * 1024 * 1024 * 1024 , (Message.MinHeaderSize, Int.MaxValue)) /* the maximum size of a single log file for some specific topic */ val logFileSizeMap = Utils.getTopicFileSize(Utils.getString(props, "topic.log.file.size" , "" )) /* the maximum time before a new log segment is rolled out */ val logRollHours = Utils.getIntInRange(props, "log.roll.hours" , 24 * 7 , ( 1 , Int.MaxValue)) /* the number of hours before rolling out a new log segment for some specific topic */ val logRollHoursMap = Utils.getTopicRollHours(Utils.getString(props, "topic.log.roll.hours" , "" )) /* the number of hours to keep a log file before deleting it */ val logRetentionHours = Utils.getIntInRange(props, "log.retention.hours" , 24 * 7 , ( 1 , Int.MaxValue)) /* the number of hours to keep a log file before deleting it for some specific topic*/ val logRetentionHoursMap = Utils.getTopicRetentionHours(Utils.getString(props, "topic.log.retention.hours" , "" )) /* the maximum size of the log before deleting it */ val logRetentionSize = Utils.getLong(props, "log.retention.size" , - 1 ) /* the maximum size of the log for some specific topic before deleting it */ val logRetentionSizeMap = Utils.getTopicRetentionSize(Utils.getString(props, "topic.log.retention.size" , "" )) /* the frequency in minutes that the log cleaner checks whether any log is eligible for deletion */ val logCleanupIntervalMinutes = Utils.getIntInRange(props, "log.cleanup.interval.mins" , 10 , ( 1 , Int.MaxValue)) /* enable zookeeper registration in the server */ val enableZookeeper = Utils.getBoolean(props, "enable.zookeeper" , true ) /* the number of messages accumulated on a log partition before messages are flushed to disk */ val flushInterval = Utils.getIntInRange(props, "log.flush.interval" , 500 , ( 1 , Int.MaxValue)) /* the maximum time in ms that a message in selected topics is kept in memory before flushed to disk, e.g., topic1:3000,topic2: 6000 */ val flushIntervalMap = Utils.getTopicFlushIntervals(Utils.getString(props, "topic.flush.intervals.ms" , "" )) /* the frequency in ms that the log flusher checks whether any log needs to be flushed to disk */ val flushSchedulerThreadRate = Utils.getInt(props, "log.default.flush.scheduler.interval.ms" , 3000 ) /* the maximum time in ms that a message in any topic is kept in memory before flushed to disk */ val defaultFlushIntervalMs = Utils.getInt(props, "log.default.flush.interval.ms" , flushSchedulerThreadRate) /* the number of partitions for selected topics, e.g., topic1:8,topic2:16 */ val topicPartitionsMap = Utils.getTopicPartitions(Utils.getString(props, "topic.partition.count.map" , "" )) /* the maximum length of topic name*/ val maxTopicNameLength = Utils.getIntInRange(props, "max.topic.name.length" , 255 , ( 1 , Int.MaxValue)) } |
上面这段代码来自kafka.server包下的KafkaConfig类，之前我们就说过，broker就是kafka中的server，所以讲配置放在这个包中也不奇怪。这里我们顺着代码往下读，也顺便看看scala的语法。和java一样也要import相关的包，kafka将同一包内的两个类写在大括号中：
| 1 | import kafka.utils.{Utils, ZKConfig} |
然后我们看类的写法：
| 1 | class KafkaConfig(props : Properties) extends ZKConfig(props) |
我们看到在加载kafkaConfig的时候会加载一个properties对象，同时也会加载有关zookeeper的properties，这个时候我们可以回忆一下，之前我们启动kafka broker的命令：
1. 启动zookeeper server ：bin/zookeeper-server-start.sh ../config/zookeeper.properties & (用&是为了能退出命令行)
2. 启动kafka server: bin/kafka-server-start.sh ../config/server.properties &
所以你能明白，初始化kafka broker的时候程序一定是去加载位于config文件夹下的properties，这个和java都一样没有区别。当然properties我们也可以通过程序来给出，这个我们后面再说，继续看我们的代码。既然找到了对应的properties文件，我们就结合代码和properties一起来看。
Kafka broker的properties中，将配置分为以下六类：
l Server Basics：关于brokerid，hostname等配置
l Socket Server Settings：关于传输的配置，端口、buffer的区间等。
l Log Basics：配置log的位置和partition的数量。
l Log Flush Policy：这部分是kafka配置中最重要的部分，决定了数据flush到disk的策略。
l Log Retention Policy：这部分主要配置日志处理时的策略。
l Zookeeper：配置zookeeper的相关信息。
在文件properties中的配置均出现在kafkaConfig这个类中，我们再看看kafkaConfig中的代码：
| 1 2 3 4 5 | /* the broker id for this server */ val brokerId : Int = Utils.getInt(props, "brokerid" ) /* the SO_SNDBUFF buffer of the socket sever sockets */ val socketSendBuffer : Int = Utils.getInt(props, "socket.send.buffer" , 100 * 1024 ) |
凡是参数中有三个的，最后一个是default，而参数只有两个的则要求你一定要配置，否则的话则报错。当然在这么多参数中肯定是有一些经验参数的，至于这些参数怎么配置我确实没有一个特别的推荐，需要在不断的测试中才能磨合出来。
当然你也可以将配置写在程序里，然后通过程序去启动broker，这样kafka的配置就可以像下面一样写：
| 1 2 3 | Properties props = new Properties(); props.setProperty( "port" , "9093" ); props.setProperty( "log.dir" , "/home/kafka/data1" ); |
我倒是觉得配置还是直接写在配置文件中比较好，如果需要修改也不会影响正在运行的服务，写在内存中，总是会有些不方便的地方。所以还是建议大家都写配置好了，后面讲到的producer和consumer都一样。
这里再提两个参数一个是brokerid，每个broker的id必须要区分；第二个参数是hostname，这个是broker和producer、consumer联系的关键，这里记住一定要改成你的地址和端口，否则永远连得都是localhost。
--------------------------------------------------------
这篇关于Kafka broker配置介绍的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
