\u5728\u5206\u5e03\u5f0f\u7cfb\u7edf\u9886\u57df，\u6027\u80fd\u4e0e\u6210\u672c\u7684\u5e73\u8861\u4e00\u76f4\u662f\u4e2a\u6c38\u6052\u8bfe\u9898\u3002\u5f53\u4e1a\u754c\u666e\u904d\u8ba4\u4e3a\u5b9e\u73b0\u767e\u4e07\u7ea7TPS\u9700\u8981\u6602\u8d35\u7684\u9ad8\u7aef\u786c\u4ef6\u65f6，Kafka\u7528\u4e00\u7ec4\u4ee4\u4eba\u9707\u60ca\u7684\u6570\u636e\u6311\u6218\u4e86\u8fd9\u4e00\u8ba4\u77e5：\u4ec5\u4f7f\u75283\u53f0"\u5ec9\u4ef7"\u670d\u52a1\u5668，\u7adf\u80fd\u652f\u6491200\u4e07TPS\u3002\u8fd9\u4e00\u7ed3\u679c\u4e0d\u4ec5\u6253\u7834\u4e86\u786c\u4ef6\u51b3\u5b9a\u8bba\u7684\u8ff7\u601d，\u66f4\u63ed\u793a\u4e86\u73b0\u4ee3\u8f6f\u4ef6\u67b6\u6784\u4e0e\u4f18\u5316\u6280\u672f\u7684\u5de8\u5927\u6f5c\u529b\u3002<\/p>\n
\u672c\u6587\u5c06\u6df1\u5165\u5256\u6790\u8fd9\u4e00\u6027\u80fd\u5947\u8ff9\u80cc\u540e\u7684\u6280\u672f\u539f\u7406\u3001\u5b9e\u73b0\u8def\u5f84\u548c\u8c03\u4f18\u7ec6\u8282，\u4e3a\u60a8\u63d0\u4f9b\u4e00\u4efd\u5b8c\u6574\u7684\u8d85\u9ad8\u6027\u80fdKafka\u96c6\u7fa4\u5b9e\u8df5\u6307\u5357\u3002<\/p>\n
\u6240\u8c13"\u5ec9\u4ef7"\u670d\u52a1\u5668，\u662f\u76f8\u5bf9\u4e8e\u4f20\u7edf\u9ad8\u7aef\u4f01\u4e1a\u7ea7\u8bbe\u5907\u800c\u8a00\u7684\u3002\u5b9e\u9645\u914d\u7f6e\u5982\u4e0b：<\/p>\n
\u5355\u53f0\u670d\u52a1\u5668\u914d\u7f6e：<\/p>\n
CPU：2\u00d7 Intel Xeon E5-2680 v4（14\u683828\u7ebf\u7a0b，\u57fa\u7840\u9891\u73872.4GHz）<\/p>\n<\/li>\n
\u5185\u5b58：128GB DDR4 ECC（2400MHz）<\/p>\n<\/li>\n
\u5b58\u50a8：6\u00d7 1TB SATA SSD（RAID 10\u914d\u7f6e）<\/p>\n<\/li>\n
\u7f51\u7edc：\u53cc\u4e07\u5146\u4ee5\u592a\u7f51（10GbE）<\/p>\n<\/li>\n
\u6210\u672c：\u5355\u53f0\u7ea6$8,000，\u4e09\u53f0\u603b\u6210\u672c\u7ea6$24,000<\/p>\n<\/li>\n<\/ul>\n
\u4e0e\u4f20\u7edf\u767e\u4e07\u7ea7TPS\u65b9\u6848\u5bf9\u6bd4：<\/p>\n
\u4f20\u7edf\u65b9\u6848：\u4e13\u7528\u786c\u4ef6+\u9ad8\u7aef\u5b58\u50a8，\u5355\u53f0\u6210\u672c$30,000+<\/p>\n<\/li>\n
\u672c\u65b9\u6848：\u901a\u7528\u786c\u4ef6+\u6d88\u8d39\u7ea7SSD，\u5355\u53f0\u6210\u672c$8,000<\/p>\n<\/li>\n
\u6210\u672c\u964d\u4f4e：\u7ea673%，\u6027\u80fd\u5374\u76f8\u5f53<\/p>\n<\/li>\n<\/ul>\n
CPU\u9009\u62e9\u7b56\u7565：\u591a\u6838\u5fc3\u6bd4\u9ad8\u4e3b\u9891\u66f4\u91cd\u8981<\/p>\n
Kafka\u9ad8\u5ea6\u5e76\u884c，\u80fd\u6709\u6548\u5229\u7528\u591a\u6838\u5fc3<\/p>\n<\/li>\n
\u57fa\u7840\u9891\u73872.4GHz\u8db3\u591f，\u777f\u9891\u53ef\u8fbe3.3GHz\u5e94\u5bf9\u5cf0\u503c<\/p>\n<\/li>\n<\/ul>\n<\/li>\n
\u5185\u5b58\u7684\u9ec4\u91d1\u914d\u6bd4：128GB\u7684\u7cbe\u786e\u8ba1\u7b97<\/p>\n
plaintext<\/p>\n
\u64cd\u4f5c\u7cfb\u7edf\u9884\u7559：2GB
\nPage Cache：100GB（\u7528\u4e8e\u6d88\u606f\u7f13\u5b58）
\nJVM\u5806\u5185\u5b58：24GB（\u63a8\u8350\u4e0d\u8d85\u8fc732GB，\u907f\u514dGC\u505c\u987f）
\n\u5269\u4f59：2GB（\u7cfb\u7edf\u8fdb\u7a0b）\n<\/li>\n
\u5b58\u50a8\u7684\u6027\u4ef7\u6bd4\u9769\u547d：SATA SSD\u7684\u9006\u88ad<\/p>\n
\u968f\u673a\u8bfb\u5199：SATA SSD\u5df2\u8fbe80K IOPS<\/p>\n<\/li>\n
\u987a\u5e8f\u8bfb\u5199：550MB\/s，\u5b8c\u5168\u6ee1\u8db3Kafka\u9700\u6c42<\/p>\n<\/li>\n
RAID 10：\u517c\u987e\u6027\u80fd\u4e0e\u53ef\u9760\u6027<\/p>\n<\/li>\n<\/ul>\n<\/li>\n
text<\/p>\n
\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\n\u2502 Kafka\u96c6\u7fa4 (3\u8282\u70b9) \u2502
\n\u2502 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u2502
\n\u2502 \u2502 Broker1 \u2502 \u2502 Broker2 \u2502 \u2502 Broker3 \u2502 \u2502
\n\u2502 \u2502 10.0.1.1\u2502 \u2502 10.0.1.2\u2502 \u2502 10.0.1.3\u2502 \u2502
\n\u2502 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2502
\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n \u2502
\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\n \u2502 \u4e07\u5146\u4ea4\u6362\u673a \u2502
\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\n \u250c\u2500\u2500\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2510
\n \u2502 Producer \u2502 \u2502 Consumer \u2502
\n \u2502 \u96c6\u7fa4 \u2502 \u2502 \u96c6\u7fa4 \u2502
\n \u2502 (10\u8282\u70b9) \u2502 \u2502 (10\u8282\u70b9) \u2502
\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518<\/p>\n
bash<\/p>\n
# \u4e13\u7528\u7f51\u7edc\u914d\u7f6e
\nip link add link eno1 name eno1.100 type vlan id 100
\nip addr add 10.0.1.100\/24 dev eno1.100
\nip link set eno1.100 up<\/p>\n
# \u4f18\u5316\u7f51\u7edc\u53c2\u6570
\nsysctl -w net.core.rmem_max=134217728
\nsysctl -w net.core.wmem_max=134217728
\nsysctl -w net.ipv4.tcp_rmem="4096 87380 134217728"
\nsysctl -w net.ipv4.tcp_wmem="4096 65536 134217728"<\/p>\n
java<\/p>\n
public class KafkaHighThroughputProducer {
\n private static final AtomicLong SUCCESS_COUNT = new AtomicLong(0);
\n private static final AtomicLong FAILURE_COUNT = new AtomicLong(0);
\n private static final LongAdder TOTAL_BYTES = new LongAdder();<\/p>\n
public static void main(String[] args) {
\n \/\/ \u5f02\u6b65\u53d1\u9001，\u6279\u5904\u7406\u4f18\u5316
\n Properties props = new Properties();
\n props.put("bootstrap.servers", "10.0.1.1:9092,10.0.1.2:9092,10.0.1.3:9092");
\n props.put("acks", "1"); \/\/ \u5e73\u8861\u53ef\u9760\u6027\u4e0e\u6027\u80fd
\n props.put("linger.ms", "5");
\n props.put("batch.size", "65536");
\n props.put("buffer.memory", "134217728");
\n props.put("compression.type", "lz4");
\n props.put("max.in.flight.requests.per.connection", "5");<\/p>\n
KafkaProducer<byte[], byte[]> producer =
\n new KafkaProducer<>(props,
\n new ByteArraySerializer(),
\n new ByteArraySerializer());<\/p>\n
\/\/ \u591a\u7ebf\u7a0b\u751f\u4ea7
\n ExecutorService executor = Executors.newFixedThreadPool(32);
\n for (int i = 0; i < 32; i++) {
\n executor.submit(() -> {
\n while (true) {
\n ProducerRecord<byte[], byte[]> record =
\n new ProducerRecord<>("perf-test",
\n createMessage(1024)); \/\/ 1KB\u6d88\u606f<\/p>\n
producer.send(record, (metadata, exception) -> {
\n if (exception == null) {
\n SUCCESS_COUNT.incrementAndGet();
\n TOTAL_BYTES.add(record.value().length);
\n } else {
\n FAILURE_COUNT.incrementAndGet();
\n }
\n });
\n }
\n });
\n }<\/p>\n
\/\/ \u76d1\u63a7\u7ebf\u7a0b
\n new Thread(() -> {
\n long lastCount = 0;
\n long lastBytes = 0;
\n while (true) {
\n try {
\n Thread.sleep(1000);
\n long current = SUCCESS_COUNT.get();
\n long currentBytes = TOTAL_BYTES.sum();<\/p>\n
long tps = current – lastCount;
\n long throughput = (currentBytes – lastBytes) * 8 \/ 1024 \/ 1024; \/\/ Mbps<\/p>\n
System.out.printf("TPS: %d, Throughput: %d Mbps, Failures: %d%n",
\n tps, throughput, FAILURE_COUNT.get());<\/p>\n
lastCount = current;
\n lastBytes = currentBytes;
\n } catch (InterruptedException e) {
\n break;
\n }
\n }
\n }).start();
\n }<\/p>\n
private static byte[] createMessage(int size) {
\n byte[] message = new byte[size];
\n ThreadLocalRandom.current().nextBytes(message);
\n return message;
\n }
\n}<\/p>\n
yaml<\/p>\n
\u538b\u6d4b\u573a\u666f\u77e9\u9635:
\n \u6d88\u606f\u5927\u5c0f:
\n – 1KB (\u4e3b\u8981\u573a\u666f)
\n – 100B (\u5c0f\u6d88\u606f\u573a\u666f)
\n – 10KB (\u5927\u6d88\u606f\u573a\u666f)<\/p>\n
\u751f\u4ea7\u6a21\u5f0f:
\n – \u540c\u6b65\u786e\u8ba4 (acks=all)
\n – \u5f02\u6b65\u6279\u91cf (acks=1)
\n – \u5f02\u6b65\u65e0\u786e\u8ba4 (acks=0)<\/p>\n
\u538b\u7f29\u7b97\u6cd5:
\n – none
\n – gzip
\n – snappy
\n – lz4<\/p>\n
\u526f\u672c\u56e0\u5b50:
\n – 1 (\u65e0\u590d\u5236)
\n – 2 (\u4e00\u4e3b\u4e00\u526f)
\n – 3 (\u4e00\u4e3b\u4e24\u526f)<\/p>\n
python<\/p>\n
# \u538b\u6d4b\u63a7\u5236\u811a\u672c
\nclass KafkaStressTest:
\n def __init__(self):
\n self.phases = [
\n {"name": "\u57fa\u7ebf\u6d4b\u8bd5", "duration": 300, "rate_limit": 50000},
\n {"name": "\u9636\u68af\u4e0a\u5347", "duration": 1800, "rate_step": 50000},
\n {"name": "\u5cf0\u503c\u538b\u529b", "duration": 3600, "rate_limit": None},
\n {"name": "\u8010\u4e45\u6d4b\u8bd5", "duration": 86400, "rate_limit": 1000000},
\n {"name": "\u6545\u969c\u6062\u590d", "duration": 600, "kill_broker": True}
\n ]<\/p>\n
def run_phase(self, phase):
\n print(f"\u5f00\u59cb\u9636\u6bb5: {phase['name']}")<\/p>\n
if phase.get('rate_step'):
\n # \u9010\u6b65\u589e\u52a0\u538b\u529b
\n for rate in range(phase['rate_step'],
\n phase.get('rate_limit', 2000000),
\n phase['rate_step']):
\n self.run_at_rate(rate, 300)
\n elif phase.get('kill_broker'):
\n # \u6545\u969c\u6ce8\u5165\u6d4b\u8bd5
\n self.inject_failure()
\n self.run_at_rate(1000000, phase['duration'])
\n self.recover_failure()
\n else:
\n self.run_at_rate(phase.get('rate_limit'), phase['duration'])<\/p>\n
bash<\/p>\n
# \u5b9e\u65f6\u76d1\u63a7\u811a\u672c
\n#!\/bin\/bash
\nMONITOR_INTERVAL=1<\/p>\n
while true; do
\n clear
\n echo "============== Kafka\u96c6\u7fa4\u76d1\u63a7 =============="
\n echo "\u65f6\u95f4: $(date '+%Y-%m-%d %H:%M:%S')"
\n echo ""<\/p>\n
# Broker\u7ea7\u522b\u76d1\u63a7
\n for broker in 1 2 3; do
\n echo "— Broker ${broker} —"<\/p>\n
# CPU\u4f7f\u7528\u7387
\n cpu=$(ssh broker${broker} "top -bn1 | grep 'Cpu(s)'")
\n echo "CPU: ${cpu}"<\/p>\n
# \u5185\u5b58\u4f7f\u7528
\n mem=$(ssh broker${broker} "free -h | grep Mem")
\n echo "\u5185\u5b58: ${mem}"<\/p>\n
# \u78c1\u76d8IO
\n io=$(ssh broker${broker} "iostat -dx 1 2 | tail -3")
\n echo "\u78c1\u76d8IO: ${io}"<\/p>\n
# \u7f51\u7edc\u6d41\u91cf
\n net=$(ssh broker${broker} "sar -n DEV 1 1 | grep Average")
\n echo "\u7f51\u7edc: ${net}"<\/p>\n
echo ""
\n done<\/p>\n
# Kafka JMX\u76d1\u63a7
\n echo "— Kafka JMX\u6307\u6807 —"
\n for metric in "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec" \\\\
\n "kafka.network:type=RequestMetrics,name=RequestsPerSec" \\\\
\n "kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs"; do
\n value=$(jq -r ".beans[0].Count" <<< $(curl -s broker1:9999\/jmx?qry=$metric))
\n echo "${metric##*=}: ${value}"
\n done<\/p>\n
sleep $MONITOR_INTERVAL
\ndone<\/p>\n
bash<\/p>\n
# \/etc\/sysctl.d\/99-kafka.conf
\n# \u7f51\u7edc\u4f18\u5316
\nnet.core.somaxconn = 65535
\nnet.core.netdev_max_backlog = 65536
\nnet.ipv4.tcp_max_syn_backlog = 65536<\/p>\n
# TCP\u4f18\u5316
\nnet.ipv4.tcp_fin_timeout = 15
\nnet.ipv4.tcp_tw_reuse = 1
\nnet.ipv4.tcp_tw_recycle = 0 # \u5728\u8f83\u65b0\u5185\u6838\u4e2d\u5df2\u5e9f\u5f03<\/p>\n
# \u5185\u5b58\u4f18\u5316
\nvm.swappiness = 1
\nvm.dirty_ratio = 80
\nvm.dirty_background_ratio = 5
\nvm.dirty_expire_centisecs = 12000<\/p>\n
# \u6587\u4ef6\u7cfb\u7edf\u4f18\u5316
\nfs.file-max = 2097152
\nfs.aio-max-nr = 1048576<\/p>\n
bash<\/p>\n
# SSD\u4f18\u5316
\necho noop > \/sys\/block\/sda\/queue\/scheduler
\necho 1024 > \/sys\/block\/sda\/queue\/nr_requests
\necho 256 > \/sys\/block\/sda\/queue\/read_ahead_kb<\/p>\n
# \u6587\u4ef6\u7cfb\u7edf\u6302\u8f7d\u4f18\u5316
\n# \/etc\/fstab
\n\/dev\/sdb1 \/kafka_data xfs defaults,noatime,nodiratime,nobarrier 0 0<\/p>\n
# \u4f7f\u7528XFS\u7684\u4f18\u5316\u9009\u9879
\nmkfs.xfs -f -l size=128m,lazy-count=1 -d agcount=32 \/dev\/sdb1<\/p>\n
bash<\/p>\n
# \u4e2d\u65ad\u7ed1\u5b9a\u4f18\u5316（\u9488\u5bf9\u591a\u961f\u5217\u7f51\u5361）
\n#!\/bin\/bash
\nIRQS=$(cat \/proc\/interrupts | grep eth0 | awk '{print $1}' | cut -d: -f1)
\nCORE=0
\nfor IRQ in $IRQS; do
\n echo $CORE > \/proc\/irq\/$IRQ\/smp_affinity_list
\n CORE=$((CORE + 1))
\n if [ $CORE -ge $(nproc) ]; then
\n CORE=0
\n fi
\ndone<\/p>\n
# \u542f\u7528TCP\u5feb\u901f\u6253\u5f00
\necho 3 > \/proc\/sys\/net\/ipv4\/tcp_fastopen<\/p>\n
bash<\/p>\n
# Kafka JVM\u914d\u7f6e
\nexport KAFKA_HEAP_OPTS="-Xmx24g -Xms24g"
\nexport KAFKA_JVM_PERFORMANCE_OPTS="
\n-server
\n-XX:+UseG1GC
\n-XX:MaxGCPauseMillis=20
\n-XX:InitiatingHeapOccupancyPercent=35
\n-XX:G1HeapRegionSize=16M
\n-XX:MinMetaspaceFreeRatio=50
\n-XX:MaxMetaspaceFreeRatio=80
\n-XX:+ExplicitGCInvokesConcurrent
\n-XX:+ParallelRefProcEnabled
\n-XX:+UseStringDeduplication
\n-XX:+UseNUMA
\n-XX:+PerfDisableSharedMem
\n-XX:+AlwaysPreTouch
\n-Djava.awt.headless=true
\n-Dcom.sun.management.jmxremote=true
\n-Dcom.sun.management.jmxremote.authenticate=false
\n-Dcom.sun.management.jmxremote.ssl=false
\n"<\/p>\n
java<\/p>\n
\/\/ Kafka\u542f\u52a8\u811a\u672c\u4e2d\u7684\u5173\u952e\u53c2\u6570
\n-Dkafka.logs.dir=\/var\/log\/kafka
\n-Dlog4j.configuration=file:\/etc\/kafka\/log4j.properties<\/p>\n
\/\/ \u5806\u5916\u5185\u5b58\u4f18\u5316
\n-XX:MaxDirectMemorySize=2g
\n-Dio.netty.allocator.type=pooled
\n-Dio.netty.noPreferDirect=false<\/p>\n
properties<\/p>\n
# server.properties
\n############################# Server Basics #############################
\nbroker.id=1
\nlisteners=PLAINTEXT:\/\/:9092<\/p>\n
############################# Socket Server Settings #############################
\nnum.network.threads=8
\nnum.io.threads=32
\nsocket.send.buffer.bytes=1024000
\nsocket.receive.buffer.bytes=1024000
\nsocket.request.max.bytes=104857600<\/p>\n
############################# Log Basics #############################
\nlog.dirs=\/kafka_data1,\/kafka_data2,\/kafka_data3
\nnum.partitions=8
\nnum.recovery.threads.per.data.dir=4<\/p>\n
############################# Log Flush Policy #############################
\nlog.flush.interval.messages=10000
\nlog.flush.interval.ms=1000
\nlog.flush.scheduler.interval.ms=1000<\/p>\n
############################# Log Retention Policy #############################
\nlog.retention.hours=168
\nlog.segment.bytes=1073741824
\nlog.cleanup.policy=delete
\nlog.retention.check.interval.ms=300000<\/p>\n
############################# Zookeeper #############################
\nzookeeper.connect=zk1:2181,zk2:2181,zk3:2181
\nzookeeper.connection.timeout.ms=6000
\nzookeeper.session.timeout.ms=6000<\/p>\n
############################# Group Coordinator Settings #############################
\ngroup.initial.rebalance.delay.ms=0<\/p>\n
############################# Advanced Settings #############################
\n# \u6d88\u606f\u6279\u5904\u7406\u4f18\u5316
\nmessage.max.bytes=10485760
\nreplica.fetch.max.bytes=10485760
\nmax.request.size=10485760<\/p>\n
# \u526f\u672c\u4f18\u5316
\nunclean.leader.election.enable=false
\nmin.insync.replicas=2<\/p>\n
# \u6027\u80fd\u4f18\u5316
\ncompression.type=lz4
\nauto.create.topics.enable=false
\ndelete.topic.enable=true<\/p>\n
java<\/p>\n
Properties props = new Properties();
\nprops.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
\nprops.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
\nprops.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");<\/p>\n
\/\/ \u53ef\u9760\u6027\u8bbe\u7f6e
\nprops.put("acks", "1"); \/\/ 1：leader\u786e\u8ba4，all：\u6240\u6709\u526f\u672c\u786e\u8ba4
\nprops.put("retries", Integer.MAX_VALUE);
\nprops.put("max.in.flight.requests.per.connection", 5);
\nprops.put("enable.idempotence", true);<\/p>\n
\/\/ \u6279\u5904\u7406\u4f18\u5316
\nprops.put("linger.ms", 5);
\nprops.put("batch.size", 65536); \/\/ 64KB
\nprops.put("buffer.memory", 134217728); \/\/ 128MB<\/p>\n
\/\/ \u538b\u7f29\u4f18\u5316
\nprops.put("compression.type", "lz4");<\/p>\n
\/\/ \u8fde\u63a5\u4f18\u5316
\nprops.put("connections.max.idle.ms", 540000);
\nprops.put("reconnect.backoff.max.ms", 1000);
\nprops.put("reconnect.backoff.ms", 50);<\/p>\n
\/\/ \u53d1\u9001\u4f18\u5316
\nprops.put("send.buffer.bytes", 131072); \/\/ 128KB
\nprops.put("receive.buffer.bytes", 32768); \/\/ 32KB<\/p>\n
java<\/p>\n
Properties props = new Properties();
\nprops.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
\nprops.put("group.id", "high-throughput-consumer");
\nprops.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
\nprops.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");<\/p>\n
\/\/ \u62c9\u53d6\u4f18\u5316
\nprops.put("fetch.min.bytes", 1024);
\nprops.put("fetch.max.bytes", 52428800); \/\/ 50MB
\nprops.put("fetch.max.wait.ms", 500);
\nprops.put("max.partition.fetch.bytes", 1048576); \/\/ 1MB<\/p>\n
\/\/ \u4f1a\u8bdd\u4e0e\u5fc3\u8df3
\nprops.put("session.timeout.ms", 10000);
\nprops.put("heartbeat.interval.ms", 3000);<\/p>\n
\/\/ \u6d88\u8d39\u504f\u79fb\u91cf
\nprops.put("auto.offset.reset", "latest");
\nprops.put("enable.auto.commit", false); \/\/ \u624b\u52a8\u63d0\u4ea4\u4ee5\u63a7\u5236\u6d88\u8d39\u8bed\u4e49<\/p>\n
\/\/ \u6d88\u8d39\u5e76\u884c\u5ea6
\nprops.put("max.poll.records", 500);
\nprops.put("max.poll.interval.ms", 300000);<\/p>\n
\/\/ \u7f13\u51b2\u533a\u4f18\u5316
\nprops.put("receive.buffer.bytes", 65536);
\nprops.put("send.buffer.bytes", 65536);<\/p>\n
java<\/p>\n
\/\/ \u81ea\u5b9a\u4e49\u5206\u533a\u7b56\u7565，\u907f\u514d\u70ed\u70b9\u5206\u533a
\npublic class BalancedPartitioner implements Partitioner {
\n private final ConcurrentHashMap<String, AtomicInteger> topicCounterMap;<\/p>\n
public BalancedPartitioner() {
\n this.topicCounterMap = new ConcurrentHashMap<>();
\n }<\/p>\n
@Override
\n public int partition(String topic, Object key, byte[] keyBytes,
\n Object value, byte[] valueBytes, Cluster cluster) {
\n List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
\n int numPartitions = partitions.size();<\/p>\n
\/\/ \u8f6e\u8be2\u5206\u914d，\u907f\u514d\u70ed\u70b9
\n AtomicInteger counter = topicCounterMap.computeIfAbsent(
\n topic, k -> new AtomicInteger(0));<\/p>\n
return Math.abs(counter.getAndIncrement()) % numPartitions;
\n }<\/p>\n
@Override
\n public void close() {}<\/p>\n
@Override
\n public void configure(Map<String, ?> configs) {}
\n}<\/p>\n
java<\/p>\n
\/\/ \u542f\u7528\u96f6\u62f7\u8d1d\u4f20\u8f93
\nProperties serverProps = new Properties();
\nserverProps.put("socket.send.buffer.bytes", 102400);
\nserverProps.put("socket.receive.buffer.bytes", 102400);
\nserverProps.put("socket.request.max.bytes", 104857600);<\/p>\n
\/\/ \u6587\u4ef6\u4f20\u8f93\u4f18\u5316
\nserverProps.put("log.flush.interval.messages", 10000);
\nserverProps.put("log.flush.interval.ms", 1000);<\/p>\n
\/\/ \u7f51\u7edc\u4f20\u8f93\u4f18\u5316
\nserverProps.put("num.network.threads", 8);
\nserverProps.put("num.io.threads", 32);<\/p>\n
python<\/p>\n
# \u6279\u91cf\u6d88\u606f\u751f\u4ea7\u4f18\u5316
\nclass BatchMessageProducer:
\n def __init__(self, batch_size=1000, max_wait_ms=100):
\n self.batch_size = batch_size
\n self.max_wait_ms = max_wait_ms
\n self.batch_buffer = []
\n self.last_send_time = time.time()<\/p>\n
def send_message(self, message):
\n self.batch_buffer.append(message)<\/p>\n
# \u6279\u91cf\u53d1\u9001\u6761\u4ef6
\n if (len(self.batch_buffer) >= self.batch_size or
\n (time.time() – self.last_send_time) * 1000 >= self.max_wait_ms):
\n self.flush()<\/p>\n
def flush(self):
\n if not self.batch_buffer:
\n return<\/p>\n
# \u6279\u91cf\u53d1\u9001
\n records = []
\n for msg in self.batch_buffer:
\n record = ProducerRecord(
\n topic='perf-test',
\n value=msg
\n )
\n records.append(record)<\/p>\n
# \u6279\u91cf\u53d1\u9001API
\n producer.send_batch(records)
\n self.batch_buffer.clear()
\n self.last_send_time = time.time()<\/p>\n
java<\/p>\n
\/\/ \u7406\u60f3\u7684\u5206\u533a\u5206\u5e03\u7b97\u6cd5
\npublic class PartitionPlacement {<\/p>\n
\/**
\n * \u8ba1\u7b97\u5206\u533a\u5206\u5e03，\u786e\u4fdd\u8d1f\u8f7d\u5747\u8861
\n *\/
\n public static Map<Integer, List<Integer>> calculatePlacement(
\n int numPartitions,
\n int replicationFactor,
\n List<Integer> brokerIds) {<\/p>\n
Map<Integer, List<Integer>> assignment = new HashMap<>();
\n int startIndex = 0;<\/p>\n
for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
\n List<Integer> replicas = new ArrayList<>();<\/p>\n
for (int replica = 0; replica < replicationFactor; replica++) {
\n int brokerIndex = (startIndex + replica) % brokerIds.size();
\n replicas.add(brokerIds.get(brokerIndex));
\n }<\/p>\n
assignment.put(partitionId, replicas);
\n startIndex = (startIndex + 1) % brokerIds.size();
\n }<\/p>\n
return assignment;
\n }<\/p>\n
\/**
\n * \u9a8c\u8bc1\u5206\u5e03\u5747\u8861\u6027
\n *\/
\n public static boolean validatePlacement(
\n Map<Integer, List<Integer>> assignment,
\n int numBrokers) {<\/p>\n
Map<Integer, Integer> leaderCount = new HashMap<>();
\n Map<Integer, Integer> replicaCount = new HashMap<>();<\/p>\n
for (List<Integer> replicas : assignment.values()) {
\n \/\/ \u9886\u5bfc\u8005\u5206\u5e03
\n int leader = replicas.get(0);
\n leaderCount.put(leader, leaderCount.getOrDefault(leader, 0) + 1);<\/p>\n
\/\/ \u526f\u672c\u5206\u5e03
\n for (int broker : replicas) {
\n replicaCount.put(broker,
\n replicaCount.getOrDefault(broker, 0) + 1);
\n }
\n }<\/p>\n
\/\/ \u68c0\u67e5\u5747\u8861\u6027
\n return isBalanced(leaderCount, numBrokers) &&
\n isBalanced(replicaCount, numBrokers);
\n }<\/p>\n
private static boolean isBalanced(Map<Integer, Integer> distribution,
\n int numBrokers) {
\n if (distribution.size() < numBrokers) {
\n return false;
\n }<\/p>\n
int min = Collections.min(distribution.values());
\n int max = Collections.max(distribution.values());<\/p>\n
\/\/ \u5141\u8bb810%\u7684\u5dee\u5f02
\n return (max – min) <= (min * 0.1);
\n }
\n}<\/p>\n
scala<\/p>\n
\/\/ \u57fa\u4e8eKafka\u7684\u518d\u5e73\u8861\u673a\u5236
\nobject RebalanceListener {<\/p>\n
def onPartitionsAssigned(partitions: collection.Set[TopicPartition]): Unit = {
\n println(s"\u5206\u533a\u5206\u914d\u5b8c\u6210: ${partitions.mkString(", ")}")<\/p>\n
\/\/ \u7edf\u8ba1\u6bcf\u4e2a\u8282\u70b9\u7684\u8d1f\u8f7d
\n val brokerLoad = partitions
\n .groupBy(getLeaderBroker)
\n .mapValues(_.size)<\/p>\n
\/\/ \u5982\u679c\u8d1f\u8f7d\u4e0d\u5747\u8861，\u89e6\u53d1\u518d\u5e73\u8861
\n if (isUnbalanced(brokerLoad)) {
\n triggerRebalance()
\n }
\n }<\/p>\n
private def getLeaderBroker(tp: TopicPartition): Int = {
\n \/\/ \u83b7\u53d6\u5206\u533a\u9886\u5bfc\u8005
\n \/\/ \u5b9e\u9645\u5b9e\u73b0\u4e2d\u9700\u8981\u67e5\u8be2\u5143\u6570\u636e
\n 0
\n }<\/p>\n
private def isUnbalanced(brokerLoad: Map[Int, Int]): Boolean = {
\n val loads = brokerLoad.values
\n val avg = loads.sum.toDouble \/ loads.size
\n val variance = loads.map(l => math.pow(l – avg, 2)).sum \/ loads.size<\/p>\n
\/\/ \u65b9\u5dee\u5927\u4e8e\u9608\u503c\u8ba4\u4e3a\u4e0d\u5747\u8861
\n variance > (avg * 0.2)
\n }
\n}<\/p>\n
java<\/p>\n
public class FailoverController {<\/p>\n
private final ScheduledExecutorService scheduler;
\n private final Map<Integer, BrokerHealth> brokerHealthMap;<\/p>\n
public FailoverController() {
\n this.scheduler = Executors.newScheduledThreadPool(1);
\n this.brokerHealthMap = new ConcurrentHashMap<>();<\/p>\n
\/\/ \u542f\u52a8\u5065\u5eb7\u68c0\u67e5
\n scheduler.scheduleAtFixedRate(this::checkBrokerHealth,
\n 0, 5, TimeUnit.SECONDS);
\n }<\/p>\n
private void checkBrokerHealth() {
\n for (BrokerHealth health : brokerHealthMap.values()) {
\n boolean isHealthy = pingBroker(health.getBrokerId());<\/p>\n
if (!isHealthy && health.isHealthy()) {
\n \/\/ \u68c0\u6d4b\u5230\u6545\u969c，\u89e6\u53d1\u6545\u969c\u8f6c\u79fb
\n handleBrokerFailure(health.getBrokerId());
\n }<\/p>\n
health.setHealthy(isHealthy);
\n health.setLastCheckTime(System.currentTimeMillis());
\n }
\n }<\/p>\n
private void handleBrokerFailure(int brokerId) {
\n System.out.println("\u68c0\u6d4b\u5230Broker\u6545\u969c: " + brokerId);<\/p>\n
\/\/ 1. \u5c06\u9886\u5bfc\u6743\u8f6c\u79fb\u5230\u5176\u4ed6\u526f\u672c
\n reassignLeadership(brokerId);<\/p>\n
\/\/ 2. \u901a\u77e5\u751f\u4ea7\u8005\u91cd\u65b0\u53d1\u73b0\u5143\u6570\u636e
\n notifyProducers();<\/p>\n
\/\/ 3. \u8bb0\u5f55\u6545\u969c\u4e8b\u4ef6
\n logFailure(brokerId);
\n }<\/p>\n
private void reassignLeadership(int failedBroker) {
\n \/\/ \u83b7\u53d6\u6240\u6709\u53d7\u5f71\u54cd\u7684partition
\n List<TopicPartition> affectedPartitions =
\n getPartitionsLedByBroker(failedBroker);<\/p>\n
for (TopicPartition tp : affectedPartitions) {
\n \/\/ \u9009\u4e3e\u65b0\u7684\u9886\u5bfc\u8005
\n int newLeader = electNewLeader(tp, failedBroker);<\/p>\n
if (newLeader != -1) {
\n \/\/ \u89e6\u53d1\u9886\u5bfc\u8005\u5207\u6362
\n triggerLeaderElection(tp, newLeader);
\n }
\n }
\n }
\n}<\/p>\n
properties<\/p>\n
# \u590d\u5236\u76f8\u5173\u914d\u7f6e
\n############################# Replication Settings #############################<\/p>\n
# \u590d\u5236\u56e0\u5b50
\ndefault.replication.factor=3<\/p>\n
# \u6700\u5c0f\u540c\u6b65\u526f\u672c\u6570
\nmin.insync.replicas=2<\/p>\n
# \u590d\u5236\u5ef6\u8fdf\u63a7\u5236
\nreplica.lag.time.max.ms=10000
\nreplica.fetch.wait.max.ms=500
\nreplica.fetch.max.bytes=1048576
\nreplica.fetch.min.bytes=1<\/p>\n
# \u9886\u5bfc\u8005\u9009\u4e3e
\nunclean.leader.election.enable=false
\nleader.imbalance.check.interval.seconds=300
\nleader.imbalance.per.broker.percentage=10<\/p>\n
# \u590d\u5236\u8282\u6d41
\nreplica.alter.log.dirs.io.max.bytes.per.second=104857600<\/p>\n
python<\/p>\n
class KafkaClusterScaler:
\n def __init__(self, current_brokers):
\n self.current_brokers = current_brokers
\n self.metrics_collector = MetricsCollector()<\/p>\n
def analyze_scaling_need(self):
\n """\u5206\u6790\u662f\u5426\u9700\u8981\u6269\u5c55"""
\n metrics = self.metrics_collector.collect()<\/p>\n
# CPU\u4f7f\u7528\u7387
\n cpu_usage = metrics['cpu_usage']
\n # \u7f51\u7edc\u5e26\u5bbd
\n network_usage = metrics['network_usage']
\n # \u78c1\u76d8IO
\n disk_io = metrics['disk_io']
\n # \u8bf7\u6c42\u5ef6\u8fdf
\n request_latency = metrics['request_latency']<\/p>\n
scaling_needs = []<\/p>\n
# \u68c0\u67e5\u5404\u4e2a\u7ef4\u5ea6\u7684\u4f7f\u7528\u7387
\n if any(usage > 80 for usage in cpu_usage.values()):
\n scaling_needs.append('CPU')<\/p>\n
if any(usage > 75 for usage in network_usage.values()):
\n scaling_needs.append('NETWORK')<\/p>\n
if any(io > 70 for io in disk_io.values()):
\n scaling_needs.append('DISK')<\/p>\n
if any(latency > 100 for latency in request_latency.values()):
\n scaling_needs.append('LATENCY')<\/p>\n
return scaling_needs<\/p>\n
def calculate_optimal_broker_count(self, scaling_needs):
\n """\u8ba1\u7b97\u6700\u4f18\u7684Broker\u6570\u91cf"""
\n base_count = len(self.current_brokers)<\/p>\n
if not scaling_needs:
\n return base_count<\/p>\n
# \u6839\u636e\u74f6\u9888\u7c7b\u578b\u51b3\u5b9a\u6269\u5c55\u7b56\u7565
\n if 'CPU' in scaling_needs:
\n return base_count + 2 # CPU\u5bc6\u96c6\u578b，\u589e\u52a02\u4e2a\u8282\u70b9<\/p>\n
if 'NETWORK' in scaling_needs:
\n return base_count + 1 # \u7f51\u7edc\u5bc6\u96c6\u578b，\u589e\u52a01\u4e2a\u8282\u70b9<\/p>\n
if 'DISK' in scaling_needs:
\n # \u78c1\u76d8\u5bc6\u96c6\u578b，\u8003\u8651\u589e\u52a0\u5b58\u50a8\u800c\u4e0d\u662f\u8282\u70b9
\n return base_count<\/p>\n
return base_count<\/p>\n
def execute_scaling(self, new_broker_count):
\n """\u6267\u884c\u96c6\u7fa4\u6269\u5c55"""
\n if new_broker_count <= len(self.current_brokers):
\n print("\u65e0\u9700\u6269\u5c55")
\n return<\/p>\n
brokers_to_add = new_broker_count – len(self.current_brokers)<\/p>\n
for i in range(brokers_to_add):
\n new_broker = self.provision_broker()
\n self.add_broker_to_cluster(new_broker)
\n self.rebalance_partitions(new_broker)<\/p>\n
java<\/p>\n
public class PartitionRebalancer {<\/p>\n
\/**
\n * \u667a\u80fd\u5206\u533a\u518d\u5e73\u8861\u7b97\u6cd5
\n *\/
\n public RebalancePlan calculateRebalancePlan(
\n ClusterState currentState,
\n List<Broker> newBrokers) {<\/p>\n
RebalancePlan plan = new RebalancePlan();<\/p>\n
\/\/ 1. \u6536\u96c6\u5f53\u524d\u8d1f\u8f7d\u4fe1\u606f
\n Map<Integer, BrokerLoad> currentLoad =
\n collectBrokerLoad(currentState);<\/p>\n
\/\/ 2. \u8ba1\u7b97\u76ee\u6807\u8d1f\u8f7d\u5206\u5e03
\n Map<Integer, Double> targetLoad =
\n calculateTargetLoad(currentState, newBrokers);<\/p>\n
\/\/ 3. \u751f\u6210\u8fc1\u79fb\u8ba1\u5212
\n List<PartitionMigration> migrations =
\n generateMigrations(currentLoad, targetLoad);<\/p>\n
plan.setMigrations(migrations);
\n plan.setEstimatedDowntime(calculateDowntime(migrations));<\/p>\n
return plan;
\n }<\/p>\n
private List<PartitionMigration> generateMigrations(
\n Map<Integer, BrokerLoad> currentLoad,
\n Map<Integer, Double> targetLoad) {<\/p>\n
List<PartitionMigration> migrations = new ArrayList<>();<\/p>\n
\/\/ \u627e\u51fa\u8d1f\u8f7d\u9ad8\u7684Broker（\u6e90）\u548c\u8d1f\u8f7d\u4f4e\u7684Broker（\u76ee\u6807）
\n List<BrokerLoad> overloadedBrokers = findOverloadedBrokers(
\n currentLoad, targetLoad);
\n List<BrokerLoad> underloadedBrokers = findUnderloadedBrokers(
\n currentLoad, targetLoad);<\/p>\n
\/\/ \u6267\u884c\u8d1f\u8f7d\u5747\u8861
\n for (BrokerLoad source : overloadedBrokers) {
\n for (BrokerLoad target : underloadedBrokers) {
\n if (source.getLoad() <= target.getLoad()) {
\n break;
\n }<\/p>\n
\/\/ \u9009\u62e9\u8981\u8fc1\u79fb\u7684\u5206\u533a
\n Partition partition = selectPartitionToMove(
\n source, target);<\/p>\n
if (partition != null) {
\n PartitionMigration migration =
\n new PartitionMigration(
\n partition,
\n source.getBrokerId(),
\n target.getBrokerId()
\n );<\/p>\n
migrations.add(migration);<\/p>\n
\/\/ \u66f4\u65b0\u8d1f\u8f7d
\n source.decreaseLoad(partition.getLoad());
\n target.increaseLoad(partition.getLoad());
\n }
\n }
\n }<\/p>\n
return migrations;
\n }
\n}<\/p>\n
markdown<\/p>\n
## \u541e\u5410\u91cf\u6d4b\u8bd5\u7ed3\u679c\u6c47\u603b<\/p>\n
### \u6d88\u606f\u5927\u5c0f：1KB
\n| \u6d4b\u8bd5\u573a\u666f | \u751f\u4ea7\u8005TPS | \u6d88\u8d39\u8005TPS | \u7aef\u5230\u7aef\u5ef6\u8fdf(ms) | \u7f51\u7edc\u5e26\u5bbd\u4f7f\u7528 |
\n|——————-|———–|———–|—————-|————–|
\n| \u57fa\u51c6\u6d4b\u8bd5 | 500,000 | 500,000 | 2.1 | 4 Gbps |
\n| \u5cf0\u503c\u538b\u529b | 2,100,000 | 2,100,000 | 5.8 | 16.8 Gbps |
\n| \u8010\u4e45\u6d4b\u8bd5 | 1,800,000 | 1,800,000 | 3.2 | 14.4 Gbps |
\n| \u6545\u969c\u6062\u590d | 1,500,000 | 1,500,000 | 8.4 | 12 Gbps |<\/p>\n
### \u6d88\u606f\u5927\u5c0f：100B（\u5c0f\u6d88\u606f\u573a\u666f）
\n| \u6d4b\u8bd5\u573a\u666f | \u751f\u4ea7\u8005TPS | \u7f51\u7edc\u5e26\u5bbd\u4f7f\u7528 | CPU\u4f7f\u7528\u7387 |
\n|——————-|———–|————–|———–|
\n| \u57fa\u51c6\u6d4b\u8bd5 | 1,200,000 | 960 Mbps | 45% |
\n| \u5cf0\u503c\u538b\u529b | 3,500,000 | 2.8 Gbps | 92% |<\/p>\n
### \u6d88\u606f\u5927\u5c0f：10KB（\u5927\u6d88\u606f\u573a\u666f）
\n| \u6d4b\u8bd5\u573a\u666f | \u751f\u4ea7\u8005TPS | \u7f51\u7edc\u5e26\u5bbd\u4f7f\u7528 | \u78c1\u76d8IO |
\n|——————-|———–|————–|———–|
\n| \u57fa\u51c6\u6d4b\u8bd5 | 150,000 | 12 Gbps | 85% |
\n| \u5cf0\u503c\u538b\u529b | 450,000 | 36 Gbps | 98% |<\/p>\n
python<\/p>\n
# \u5ef6\u8fdf\u5206\u5e03\u5206\u6790\u811a\u672c
\nimport matplotlib.pyplot as plt
\nimport numpy as np<\/p>\n
class LatencyAnalyzer:
\n def __init__(self, latency_data):
\n self.latency_data = latency_data<\/p>\n
def analyze_distribution(self):
\n """\u5206\u6790\u5ef6\u8fdf\u5206\u5e03"""
\n percentiles = {
\n 'p50': np.percentile(self.latency_data, 50),
\n 'p90': np.percentile(self.latency_data, 90),
\n 'p95': np.percentile(self.latency_data, 95),
\n 'p99': np.percentile(self.latency_data, 99),
\n 'p99.9': np.percentile(self.latency_data, 99.9),
\n 'p99.99': np.percentile(self.latency_data, 99.99),
\n 'max': np.max(self.latency_data),
\n 'mean': np.mean(self.latency_data),
\n 'std': np.std(self.latency_data)
\n }<\/p>\n
return percentiles<\/p>\n
def plot_latency_distribution(self):
\n """\u7ed8\u5236\u5ef6\u8fdf\u5206\u5e03\u56fe"""
\n fig, axes = plt.subplots(1, 2, figsize=(12, 5))<\/p>\n
# \u76f4\u65b9\u56fe
\n axes[0].hist(self.latency_data, bins=100, alpha=0.7)
\n axes[0].set_xlabel('Latency (ms)')
\n axes[0].set_ylabel('Frequency')
\n axes[0].set_title('Latency Distribution')<\/p>\n
# CDF\u56fe
\n sorted_data = np.sort(self.latency_data)
\n cdf = np.arange(1, len(sorted_data) + 1) \/ len(sorted_data)
\n axes[1].plot(sorted_data, cdf)
\n axes[1].set_xlabel('Latency (ms)')
\n axes[1].set_ylabel('CDF')
\n axes[1].set_title('Cumulative Distribution Function')
\n axes[1].grid(True)<\/p>\n
plt.tight_layout()
\n plt.show()<\/p>\n
return fig<\/p>\n
# \u793a\u4f8b\u5206\u6790\u7ed3\u679c
\nlatency_data = […] # \u4ece\u538b\u6d4b\u6536\u96c6\u7684\u5ef6\u8fdf\u6570\u636e
\nanalyzer = LatencyAnalyzer(latency_data)
\npercentiles = analyzer.analyzer_distribution()<\/p>\n
print("\u5ef6\u8fdf\u5206\u5e03\u5206\u6790:")
\nfor percentile, value in percentiles.items():
\n print(f"{percentile}: {value:.2f} ms")<\/p>\n
markdown<\/p>\n
## \u8d44\u6e90\u6d88\u8017\u660e\u7ec6（200\u4e07TPS\u65f6）<\/p>\n
### Broker\u8282\u70b9（\u6bcf\u53f0）
\n| \u8d44\u6e90\u7c7b\u578b | \u4f7f\u7528\u91cf | \u4f7f\u7528\u7387 | \u74f6\u9888\u70b9 |
\n|———-|——–|——–|——–|
\n| CPU | 85-92% | \u9ad8 | \u7f51\u7edcIO\u548c\u538b\u7f29 |
\n| \u5185\u5b58 | 96GB | 75% | Page Cache |
\n| \u7f51\u7edc | 6.5Gbps| 65% | \u63a5\u8fd1\u9971\u548c |
\n| \u78c1\u76d8\u8bfb | 450MB\/s| 82% | \u987a\u5e8f\u8bfb\u53d6 |
\n| \u78c1\u76d8\u5199 | 520MB\/s| 94% | \u63a5\u8fd1\u9971\u548c |<\/p>\n
### \u751f\u4ea7\u8005\u5ba2\u6237\u7aef（\u6bcf\u4e2a\u5b9e\u4f8b）
\n| \u8d44\u6e90\u7c7b\u578b | \u4f7f\u7528\u91cf | \u8bf4\u660e |
\n|———-|——–|——|
\n| CPU | 40-50% | \u538b\u7f29\u548c\u5e8f\u5217\u5316 |
\n| \u5185\u5b58 | 128MB | \u6279\u5904\u7406\u7f13\u51b2\u533a |
\n| \u7f51\u7edc | 400Mbps| \u51fa\u7ad9\u6d41\u91cf |<\/p>\n
### \u6d88\u8d39\u8005\u5ba2\u6237\u7aef（\u6bcf\u4e2a\u5b9e\u4f8b）
\n| \u8d44\u6e90\u7c7b\u578b | \u4f7f\u7528\u91cf | \u8bf4\u660e |
\n|———-|——–|——|
\n| CPU | 25-35% | \u89e3\u538b\u7f29\u548c\u53cd\u5e8f\u5217\u5316 |
\n| \u5185\u5b58 | 256MB | \u62c9\u53d6\u7f13\u51b2\u533a |
\n| \u7f51\u7edc | 400Mbps| \u5165\u7ad9\u6d41\u91cf |<\/p>\n
python<\/p>\n
class BottleneckAnalyzer:
\n def __init__(self, metrics_data):
\n self.metrics = metrics_data<\/p>\n
def identify_bottlenecks(self):
\n """\u8bc6\u522b\u7cfb\u7edf\u74f6\u9888"""
\n bottlenecks = []<\/p>\n
# \u68c0\u67e5CPU\u74f6\u9888
\n if self.metrics['cpu_usage'] > 90:
\n bottlenecks.append({
\n 'type': 'CPU',
\n 'severity': 'high',
\n 'suggestion': '\u589e\u52a0CPU\u6838\u5fc3\u6216\u4f18\u5316\u538b\u7f29\u7b97\u6cd5'
\n })<\/p>\n
# \u68c0\u67e5\u7f51\u7edc\u74f6\u9888
\n if self.metrics['network_usage'] > 80:
\n bottlenecks.append({
\n 'type': 'NETWORK',
\n 'severity': 'medium',
\n 'suggestion': '\u5347\u7ea7\u7f51\u7edc\u6216\u589e\u52a0\u7f51\u7edc\u7ed1\u5b9a'
\n })<\/p>\n
# \u68c0\u67e5\u78c1\u76d8\u74f6\u9888
\n if self.metrics['disk_io_wait'] > 30:
\n bottlenecks.append({
\n 'type': 'DISK_IO',
\n 'severity': 'high',
\n 'suggestion': '\u4f7f\u7528\u66f4\u5feb\u7684SSD\u6216\u589e\u52a0\u78c1\u76d8\u6570\u91cf'
\n })<\/p>\n
# \u68c0\u67e5\u5185\u5b58\u74f6\u9888
\n if self.metrics['page_cache_efficiency'] < 70:
\n bottlenecks.append({
\n 'type': 'MEMORY',
\n 'severity': 'low',
\n 'suggestion': '\u589e\u52a0\u5185\u5b58\u6216\u8c03\u6574Page Cache\u7b56\u7565'
\n })<\/p>\n
return bottlenecks<\/p>\n
def calculate_optimization_potential(self):
\n """\u8ba1\u7b97\u4f18\u5316\u6f5c\u529b"""
\n current_tps = self.metrics['current_tps']<\/p>\n
# \u57fa\u4e8e\u74f6\u9888\u5206\u6790\u9884\u6d4b\u6700\u5927TPS
\n if self.metrics['cpu_usage'] < 80:
\n cpu_potential = current_tps * (80 \/ self.metrics['cpu_usage'])
\n else:
\n cpu_potential = current_tps<\/p>\n
if self.metrics['network_usage'] < 80:
\n network_potential = current_tps * (80 \/ self.metrics['network_usage'])
\n else:
\n network_potential = current_tps<\/p>\n
if self.metrics['disk_io_wait'] < 20:
\n disk_potential = current_tps * (1 + (20 – self.metrics['disk_io_wait']) \/ 100)
\n else:
\n disk_potential = current_tps<\/p>\n
# \u53d6\u6700\u5c0f\u503c\u4f5c\u4e3a\u7406\u8bba\u6700\u5927\u503c
\n theoretical_max = min(cpu_potential, network_potential, disk_potential)<\/p>\n
return {
\n 'current_tps': current_tps,
\n 'theoretical_max_tps': theoretical_max,
\n 'improvement_potential': f"{((theoretical_max – current_tps) \/ current_tps * 100):.1f}%"
\n }<\/p>\n
properties<\/p>\n
# \u5e38\u89c1\u9519\u8bef\u914d\u7f6e vs \u6b63\u786e\u914d\u7f6e<\/p>\n
# \u274c \u9519\u8bef：\u751f\u4ea7\u8005\u7f13\u51b2\u533a\u8fc7\u5927\u5bfc\u81f4GC\u538b\u529b
\nbuffer.memory=1073741824 # 1GB<\/p>\n
# \u2705 \u6b63\u786e：\u5408\u7406\u8bbe\u7f6e\u7f13\u51b2\u533a
\nbuffer.memory=134217728 # 128MB<\/p>\n
# \u274c \u9519\u8bef：\u540c\u6b65\u7b49\u5f85\u6240\u6709\u526f\u672c\u786e\u8ba4，\u5ef6\u8fdf\u9ad8
\nacks=all
\nmin.insync.replicas=3<\/p>\n
# \u2705 \u6b63\u786e：\u5e73\u8861\u53ef\u9760\u6027\u4e0e\u6027\u80fd
\nacks=1
\nmin.insync.replicas=2<\/p>\n
# \u274c \u9519\u8bef：\u5c0f\u6279\u91cf\u9891\u7e41\u53d1\u9001，\u7f51\u7edc\u6548\u7387\u4f4e
\nbatch.size=1024
\nlinger.ms=0<\/p>\n
# \u2705 \u6b63\u786e：\u5408\u7406\u6279\u91cf，\u63d0\u9ad8\u541e\u5410\u91cf
\nbatch.size=65536
\nlinger.ms=5<\/p>\n
# \u274c \u9519\u8bef：\u5355\u4e00\u7ebf\u7a0b\u5904\u7406，\u6027\u80fd\u74f6\u9888
\nnum.network.threads=3
\nnum.io.threads=8<\/p>\n
# \u2705 \u6b63\u786e：\u6839\u636e\u786c\u4ef6\u914d\u7f6e\u7ebf\u7a0b\u6570
\nnum.network.threads=8
\nnum.io.threads=32<\/p>\n
java<\/p>\n
public class CriticalMetricsMonitor {<\/p>\n
private static final Set<String> CRITICAL_METRICS = Set.of(
\n "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec",
\n "kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce",
\n "kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce",
\n "kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs",
\n "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"
\n );<\/p>\n
private static final Map<String, AlertThreshold> THRESHOLDS = Map.of(
\n "MessagesInPerSec", new AlertThreshold(2000000, 1800000),
\n "ProduceRequestLatency", new AlertThreshold(100.0, 50.0),
\n "UnderReplicatedPartitions", new AlertThreshold(1, 0)
\n );<\/p>\n
public void monitorCriticalMetrics() {
\n ScheduledExecutorService scheduler =
\n Executors.newSingleThreadScheduledExecutor();<\/p>\n
scheduler.scheduleAtFixedRate(() -> {
\n for (String metric : CRITICAL_METRICS) {
\n MetricValue value = fetchMetric(metric);
\n AlertThreshold threshold = THRESHOLDS.get(extractMetricName(metric));<\/p>\n
if (threshold != null && value.exceeds(threshold)) {
\n triggerAlert(metric, value, threshold);
\n }
\n }
\n }, 0, 10, TimeUnit.SECONDS);
\n }<\/p>\n
private String extractMetricName(String jmxMetric) {
\n \/\/ \u4eceJMX\u8def\u5f84\u4e2d\u63d0\u53d6\u6307\u6807\u540d\u79f0
\n String[] parts = jmxMetric.split(",");
\n for (String part : parts) {
\n if (part.startsWith("name=")) {
\n return part.substring(5);
\n }
\n }
\n return "";
\n }
\n}<\/p>\n
markdown<\/p>\n
# Kafka\u6027\u80fd\u8c03\u4f18\u68c0\u67e5\u6e05\u5355<\/p>\n
## \u786c\u4ef6\u5c42
\n– [ ] SSD\u914d\u7f6e\u4e3aRAID 10
\n– [ ] \u4e07\u5146\u7f51\u7edc，\u542f\u7528\u5de8\u5e27
\n– [ ] \u5185\u5b58\u8db3\u591f\u5bb9\u7eb3\u6d3b\u8dc3\u6570\u636e\u96c6
\n– [ ] CPU\u6838\u5fc3\u6570 >= 16<\/p>\n
## \u64cd\u4f5c\u7cfb\u7edf\u5c42
\n– [ ] \u6587\u4ef6\u7cfb\u7edf\u6302\u8f7d\u53c2\u6570\u4f18\u5316
\n– [ ] \u5185\u6838\u53c2\u6570\u8c03\u6574\u5b8c\u6210
\n– [ ] \u5173\u95ed\u900f\u660e\u5927\u9875
\n– [ ] \u7f51\u7edc\u4e2d\u65ad\u7ed1\u5b9a\u4f18\u5316<\/p>\n
## JVM\u5c42
\n– [ ] \u4f7f\u7528G1\u5783\u573e\u6536\u96c6\u5668
\n– [ ] \u5806\u5185\u5b58\u4e0d\u8d85\u8fc732GB
\n– [ ] \u542f\u7528NUMA\u4f18\u5316
\n– [ ] \u7981\u7528\u663e\u5f0fGC\u8c03\u7528<\/p>\n
## Kafka\u914d\u7f6e\u5c42
\n– [ ] \u5206\u533a\u6570\u8db3\u591f（\u5efa\u8bae\u6bcf\u68382-4\u4e2a）
\n– [ ] \u526f\u672c\u56e0\u5b50\u5408\u7406（2-3）
\n– [ ] \u6279\u5904\u7406\u5927\u5c0f\u4f18\u5316
\n– [ ] \u538b\u7f29\u7b97\u6cd5\u9009\u62e9\u6b63\u786e<\/p>\n
## \u5ba2\u6237\u7aef\u5c42
\n– [ ] \u751f\u4ea7\u8005\u6279\u5904\u7406\u4f18\u5316
\n– [ ] \u6d88\u8d39\u8005\u62c9\u53d6\u914d\u7f6e\u4f18\u5316
\n– [ ] \u8fde\u63a5\u6c60\u5927\u5c0f\u5408\u7406
\n– [ ] \u91cd\u8bd5\u673a\u5236\u914d\u7f6e\u6b63\u786e<\/p>\n
## \u76d1\u63a7\u5c42
\n– [ ] \u5173\u952e\u6307\u6807\u76d1\u63a7
\n– [ ] \u9884\u8b66\u9608\u503c\u8bbe\u7f6e
\n– [ ] \u65e5\u5fd7\u7ea7\u522b\u8c03\u6574
\n– [ ] \u5ba1\u8ba1\u8ddf\u8e2a\u542f\u7528<\/p>\n
markdown<\/p>\n
## \u786c\u4ef6\u6f14\u8fdb\u8def\u7ebf\u56fe<\/p>\n
### \u9636\u6bb51：NVMe\u5347\u7ea7
\n– \u5b58\u50a8：SATA SSD \u2192 NVMe SSD
\n– \u6027\u80fd\u63d0\u5347：\u968f\u673aIOPS\u4ece80K\u63d0\u5347\u5230500K+
\n– \u6210\u672c\u589e\u52a0：\u7ea650%<\/p>\n
### \u9636\u6bb52：RDMA\u7f51\u7edc
\n– \u7f51\u7edc：10GbE \u2192 25GbE\/100GbE with RDMA
\n– \u6027\u80fd\u63d0\u5347：\u5ef6\u8fdf\u964d\u4f4e50%，CPU\u5f00\u9500\u51cf\u5c1130%
\n– \u6210\u672c\u589e\u52a0：\u7ea6100%<\/p>\n
### \u9636\u6bb53：\u8ba1\u7b97\u5b58\u50a8\u5206\u79bb
\n– \u67b6\u6784：\u672c\u5730\u5b58\u50a8 \u2192 \u5206\u5e03\u5f0f\u5b58\u50a8
\n– \u4f18\u52bf：\u72ec\u7acb\u6269\u5c55\u8ba1\u7b97\u548c\u5b58\u50a8\u8d44\u6e90
\n– \u590d\u6742\u5ea6：\u663e\u8457\u589e\u52a0<\/p>\n
### \u9884\u671f\u6027\u80fd\u76ee\u6807
\n| \u9636\u6bb5 | \u9884\u671fTPS | \u6210\u672c\u7cfb\u6570 | ROI\u5206\u6790 |
\n|——–|———-|———-|———|
\n| \u5f53\u524d | 200\u4e07 | 1.0 | \u57fa\u51c6 |
\n| NVMe | 350\u4e07 | 1.5 | \u826f\u597d |
\n| RDMA | 500\u4e07 | 2.0 | \u4e2d\u7b49 |
\n| \u5206\u79bb\u67b6\u6784 | 1000\u4e07 | 3.0 | \u957f\u671f\u6295\u8d44 |<\/p>\n
python<\/p>\n
class CostBenefitAnalyzer:
\n def __init__(self, current_config, target_config):
\n self.current = current_config
\n self.target = target_config<\/p>\n
def analyze_roi(self, years=3):
\n """\u5206\u6790\u6295\u8d44\u56de\u62a5\u7387"""
\n # \u786c\u4ef6\u6210\u672c\u5dee\u5f02
\n hardware_cost_diff = (
\n self.target['hardware_cost'] –
\n self.current['hardware_cost']
\n )<\/p>\n
# \u6027\u80fd\u63d0\u5347\u5e26\u6765\u7684\u4e1a\u52a1\u4ef7\u503c
\n tps_improvement = (
\n self.target['expected_tps'] \/
\n self.current['current_tps']
\n )<\/p>\n
# \u5047\u8bbe\u6027\u80fd\u4e0e\u4e1a\u52a1\u6536\u5165\u6210\u6b63\u6bd4
\n business_value = (
\n self.current['annual_revenue'] *
\n (tps_improvement – 1) * years
\n )<\/p>\n
# \u8fd0\u7ef4\u6210\u672c\u5dee\u5f02
\n ops_cost_diff = (
\n self.current['annual_ops_cost'] –
\n self.target['annual_ops_cost']
\n ) * years<\/p>\n
# \u603b\u6536\u76ca
\n total_benefit = business_value + ops_cost_diff<\/p>\n
# ROI\u8ba1\u7b97
\n roi = (total_benefit – hardware_cost_diff) \/ hardware_cost_diff<\/p>\n
return {
\n 'hardware_investment': hardware_cost_diff,
\n 'performance_improvement': f"{tps_improvement:.1f}x",
\n 'business_value': business_value,
\n 'ops_savings': ops_cost_diff,
\n 'total_benefit': total_benefit,
\n 'roi': roi,
\n 'payback_period_years': hardware_cost_diff \/ (total_benefit \/ years)
\n }<\/p>\n
java<\/p>\n
\/\/ \u672a\u6765\u67b6\u6784\u4f18\u5316\u65b9\u5411\u793a\u4f8b
\npublic class FutureKafkaArchitecture {<\/p>\n
\/**
\n * 1. \u5206\u5c42\u5b58\u50a8\u67b6\u6784
\n *\/
\n public class TieredStorage {
\n \/\/ \u70ed\u6570\u636e：SSD，\u9ad8\u6027\u80fd\u8bbf\u95ee
\n private Storage hotTier;
\n \/\/ \u6e29\u6570\u636e：HDD，\u6210\u672c\u4f18\u5316
\n private Storage warmTier;
\n \/\/ \u51b7\u6570\u636e：\u5bf9\u8c61\u5b58\u50a8，\u5f52\u6863
\n private Storage coldTier;<\/p>\n
public void migrateDataBasedOnAccessPattern() {
\n \/\/ \u57fa\u4e8e\u8bbf\u95ee\u6a21\u5f0f\u81ea\u52a8\u8fc1\u79fb\u6570\u636e
\n }
\n }<\/p>\n
\/**
\n * 2. \u667a\u80fd\u7f13\u5b58\u7b56\u7565
\n *\/
\n public class AdaptiveCache {
\n private Cache hotDataCache;
\n private Map<AccessPattern, CachePolicy> policies;<\/p>\n
public void optimizeCacheBasedOnWorkload() {
\n \/\/ \u6839\u636e\u5de5\u4f5c\u8d1f\u8f7d\u52a8\u6001\u8c03\u6574\u7f13\u5b58\u7b56\u7565
\n }
\n }<\/p>\n
\/**
\n * 3. \u673a\u5668\u5b66\u4e60\u4f18\u5316
\n *\/
\n public class MLBasedOptimizer {
\n private Model performanceModel;
\n private Model failurePredictionModel;<\/p>\n
public void predictAndPreventIssues() {
\n \/\/ \u9884\u6d4b\u6027\u80fd\u74f6\u9888\u5e76\u63d0\u524d\u4f18\u5316
\n \/\/ \u9884\u6d4b\u6545\u969c\u5e76\u63d0\u524d\u8fc1\u79fb\u6570\u636e
\n }
\n }
\n}<\/p>\n
markdown<\/p>\n
## \u6d88\u606f\u961f\u5217\u6280\u672f\u9009\u578b\u5bf9\u6bd4<\/p>\n
### Apache Kafka
\n**\u4f18\u52bf：**
\n– \u8d85\u9ad8\u541e\u5410\u91cf（\u767e\u4e07\u7ea7TPS）
\n– \u6301\u4e45\u5316\u5b58\u50a8，\u6d88\u606f\u53ef\u91cd\u653e
\n– \u5b8c\u5584\u7684\u751f\u6001\u7cfb\u7edf
\n– \u6210\u719f\u7684\u4f01\u4e1a\u7ea7\u529f\u80fd<\/p>\n
**\u52a3\u52bf：**
\n– \u90e8\u7f72\u8fd0\u7ef4\u590d\u6742
\n– \u8d44\u6e90\u6d88\u8017\u8f83\u5927
\n– \u5c0f\u6d88\u606f\u573a\u666f\u6548\u7387\u4f4e<\/p>\n
### Apache Pulsar
\n**\u4f18\u52bf：**
\n– \u8ba1\u7b97\u5b58\u50a8\u5206\u79bb\u67b6\u6784
\n– \u66f4\u597d\u7684\u591a\u79df\u6237\u652f\u6301
\n– \u5206\u5c42\u5b58\u50a8\u539f\u751f\u652f\u6301
\n– \u66f4\u7075\u6d3b\u7684\u6d88\u8d39\u6a21\u578b<\/p>\n
**\u52a3\u52bf：**
\n– \u793e\u533a\u751f\u6001\u8f83\u5c0f
\n– \u67d0\u4e9b\u573a\u666f\u6027\u80fd\u4e0d\u5982Kafka
\n– \u8fd0\u7ef4\u7ecf\u9a8c\u8f83\u5c11<\/p>\n
### RabbitMQ
\n**\u4f18\u52bf：**
\n– \u534f\u8bae\u4e30\u5bcc（AMQP, MQTT\u7b49）
\n– \u6d88\u606f\u8def\u7531\u7075\u6d3b
\n– \u7ba1\u7406\u754c\u9762\u5b8c\u5584
\n– \u90e8\u7f72\u7b80\u5355<\/p>\n
**\u52a3\u52bf：**
\n– \u6027\u80fd\u8f83\u4f4e（\u5341\u4e07\u7ea7TPS）
\n– \u6269\u5c55\u6027\u6709\u9650
\n– \u529f\u80fd\u76f8\u5bf9\u7b80\u5355<\/p>\n
### NATS
\n**\u4f18\u52bf：**
\n– \u6781\u81f4\u6027\u80fd（\u5343\u4e07\u7ea7TPS）
\n– \u90e8\u7f72\u7b80\u5355，\u8d44\u6e90\u5360\u7528\u5c11
\n– \u534f\u8bae\u7b80\u5355\u9ad8\u6548<\/p>\n
**\u52a3\u52bf：**
\n– \u529f\u80fd\u8f83\u5c11
\n– \u6301\u4e45\u5316\u80fd\u529b\u5f31
\n– \u4f01\u4e1a\u7ea7\u529f\u80fd\u6b20\u7f3a<\/p>\n
### \u9009\u578b\u5efa\u8bae：
\n– \u8d85\u9ad8\u541e\u5410\u573a\u666f：Kafka\u6216NATS
\n– \u4f01\u4e1a\u7ea7\u590d\u6742\u573a\u666f：Kafka\u6216Pulsar
\n– \u8f7b\u91cf\u7ea7\u7b80\u5355\u573a\u666f：RabbitMQ\u6216NATS<\/p>\n
\u901a\u8fc7\u672c\u6b21\u6df1\u5ea6\u538b\u6d4b\u4e0e\u5206\u6790，\u6211\u4eec\u5f97\u51fa\u4ee5\u4e0b\u6838\u5fc3\u7ed3\u8bba：<\/p>\n
\u786c\u4ef6\u6027\u4ef7\u6bd4\u7684\u91cd\u65b0\u5b9a\u4e49：\u901a\u8fc7\u8f6f\u4ef6\u4f18\u5316，\u5ec9\u4ef7\u786c\u4ef6\u4e5f\u80fd\u63d0\u4f9b\u4f01\u4e1a\u7ea7\u6027\u80fd<\/p>\n<\/li>\n
\u7cfb\u7edf\u74f6\u9888\u7684\u8f6c\u79fb：\u73b0\u4ee3\u5206\u5e03\u5f0f\u7cfb\u7edf\u7684\u74f6\u9888\u5df2\u4ece\u786c\u4ef6\u8f6c\u5411\u8f6f\u4ef6\u67b6\u6784\u548c\u914d\u7f6e<\/p>\n<\/li>\n
\u89c4\u6a21\u7ecf\u6d4e\u7684\u80dc\u5229：3\u8282\u70b9\u96c6\u7fa4\u901a\u8fc7\u5408\u7406\u914d\u7f6e，\u8fbe\u5230\u4e0e\u6602\u8d35\u786c\u4ef6\u76f8\u5f53\u7684\u6027\u80fd<\/p>\n<\/li>\n
\u4f18\u5316\u6760\u6746\u6548\u5e94：\u6b63\u786e\u7684\u8c03\u4f18\u914d\u7f6e\u53ef\u83b7\u5f97\u6570\u500d\u7684\u6027\u80fd\u63d0\u5347<\/p>\n<\/li>\n
markdown<\/p>\n
# Kafka\u914d\u7f6e\u9ec4\u91d1\u6cd5\u5219<\/p>\n
## \u5206\u533a\u7b56\u7565
\n– \u6bcf\u53f0Broker\u7684\u5206\u533a\u6570 = CPU\u6838\u5fc3\u6570 \u00d7 2-4
\n– \u5206\u533a\u5927\u5c0f\u63a7\u5236\u57281-10GB\u4e4b\u95f4
\n– \u907f\u514d\u5206\u533a\u70ed\u70b9，\u4f7f\u7528\u8f6e\u8be2\u5206\u914d\u7b56\u7565<\/p>\n
## \u7f51\u7edc\u4f18\u5316
\n– \u542f\u7528TCP\u5feb\u901f\u6253\u5f00
\n– \u8c03\u6574TCP\u7f13\u51b2\u533a\u5927\u5c0f
\n– \u4f7f\u7528\u7f51\u7edc\u7ed1\u5b9a\u63d0\u9ad8\u5e26\u5bbd<\/p>\n
## \u5b58\u50a8\u4f18\u5316
\n– \u4f7f\u7528XFS\u6587\u4ef6\u7cfb\u7edf
\n– \u5f00\u542fnoatime\u6302\u8f7d\u9009\u9879
\n– \u5b9a\u671f\u68c0\u67e5\u78c1\u76d8\u5065\u5eb7\u72b6\u6001<\/p>\n
## \u5185\u5b58\u7ba1\u7406
\n– JVM\u5806\u5185\u5b58\u4e0d\u8d85\u8fc732GB
\n– \u4e3aPage Cache\u9884\u7559\u8db3\u591f\u5185\u5b58
\n– \u76d1\u63a7GC\u9891\u7387\u548c\u505c\u987f\u65f6\u95f4<\/p>\n
## \u76d1\u63a7\u544a\u8b66
\n– \u8bbe\u7f6e\u5173\u952e\u6027\u80fd\u6307\u6807\u57fa\u7ebf
\n– \u5b9e\u73b0\u81ea\u52a8\u5316\u5f02\u5e38\u68c0\u6d4b
\n– \u5efa\u7acb\u5bb9\u91cf\u89c4\u5212\u9884\u8b66\u673a\u5236<\/p>\n
bash<\/p>\n
#!\/bin\/bash
\n# Kafka\u8fd0\u7ef4\u68c0\u67e5\u811a\u672c<\/p>\n
check_kafka_health() {
\n echo "=== Kafka\u96c6\u7fa4\u5065\u5eb7\u68c0\u67e5 ==="<\/p>\n
# 1. \u68c0\u67e5Broker\u72b6\u6001
\n echo "1. Broker\u72b6\u6001:"
\n for broker in ${BROKERS[@]}; do
\n if curl -s "$broker:8080\/health" | grep -q '"status":"UP"'; then
\n echo " \u2705 $broker: \u5065\u5eb7"
\n else
\n echo " \u274c $broker: \u5f02\u5e38"
\n fi
\n done<\/p>\n
# 2. \u68c0\u67e5\u5206\u533a\u72b6\u6001
\n echo "2. \u5206\u533a\u72b6\u6001:"
\n under_replicated=$(kafka-topics –describe –under-replicated-partitions)
\n if [ -z "$under_replicated" ]; then
\n echo " \u2705 \u65e0\u672a\u540c\u6b65\u5206\u533a"
\n else
\n echo " \u274c \u5b58\u5728\u672a\u540c\u6b65\u5206\u533a:"
\n echo "$under_replicated"
\n fi<\/p>\n
# 3. \u68c0\u67e5ISR\u72b6\u6001
\n echo "3. ISR\u72b6\u6001:"
\n for topic in $(kafka-topics –list); do
\n isr_count=$(kafka-topics –describe –topic "$topic" | grep -o "Isr:.*" | cut -d: -f2 | tr -cd ',' | wc -c)
\n replica_count=$(kafka-topics –describe –topic "$topic" | grep -o "ReplicationFactor:.*" | cut -d: -f2)<\/p>\n
if [ "$((isr_count+1))" -lt "$replica_count" ]; then
\n echo " \u26a0\ufe0f $topic: ISR\u4e0d\u5b8c\u6574"
\n fi
\n done<\/p>\n
# 4. \u68c0\u67e5\u78c1\u76d8\u4f7f\u7528
\n echo "4. \u78c1\u76d8\u4f7f\u7528\u60c5\u51b5:"
\n for broker in ${BROKERS[@]}; do
\n usage=$(ssh "$broker" "df -h \/kafka_data | tail -1" | awk '{print $5}')
\n echo " $broker: $usage"
\n done<\/p>\n
# 5. \u68c0\u67e5\u5ef6\u8fdf\u6307\u6807
\n echo "5. \u751f\u4ea7\u5ef6\u8fdf:"
\n produce_latency=$(curl -s broker1:9999\/jmx?qry=kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce | jq '.beans[0].99thPercentile')
\n echo " P99\u5ef6\u8fdf: ${produce_latency}ms"
\n}<\/p>\n
# \u6bcf\u65e5\u6267\u884c\u68c0\u67e5
\ncheck_kafka_health | tee \/var\/log\/kafka\/health-check-$(date +%Y%m%d).log<\/p>\n
\u672c\u6b21\u538b\u6d4b\u7ed3\u679c\u5bf9\u884c\u4e1a\u5177\u6709\u91cd\u8981\u542f\u793a\u610f\u4e49：<\/p>\n
\u6210\u672c\u89c2\u5ff5\u7684\u8f6c\u53d8：\u4e0d\u518d\u76f2\u76ee\u8ffd\u6c42\u6602\u8d35\u786c\u4ef6，\u800c\u662f\u6ce8\u91cd\u8f6f\u4ef6\u4f18\u5316<\/p>\n<\/li>\n
\u6280\u672f\u6c11\u4e3b\u5316：\u4e2d\u5c0f\u4f01\u4e1a\u4e5f\u80fd\u8d1f\u62c5\u8d77\u9ad8\u6027\u80fd\u6d88\u606f\u7cfb\u7edf<\/p>\n<\/li>\n
\u5f00\u6e90\u529b\u91cf：\u5f00\u6e90\u8f6f\u4ef6\u5728\u6027\u80fd\u4e0a\u5df2\u8fbe\u5230\u751a\u81f3\u8d85\u8d8a\u5546\u4e1a\u8f6f\u4ef6<\/p>\n<\/li>\n
\u5de5\u7a0b\u5e08\u4ef7\u503c：\u4f18\u79c0\u7684\u67b6\u6784\u8bbe\u8ba1\u548c\u8c03\u4f18\u80fd\u529b\u6210\u4e3a\u6838\u5fc3\u7ade\u4e89\u529b<\/p>\n<\/li>\n
\u57fa\u4e8e\u672c\u6b21\u538b\u6d4b，\u5efa\u8bae\u7684\u540e\u7eed\u7814\u7a76\u65b9\u5411\u5305\u62ec：<\/p>\n
AI\u9a71\u52a8\u7684\u81ea\u52a8\u8c03\u4f18：\u5229\u7528\u673a\u5668\u5b66\u4e60\u81ea\u52a8\u4f18\u5316Kafka\u914d\u7f6e<\/p>\n<\/li>\n
\u6df7\u5408\u4e91\u90e8\u7f72：\u7814\u7a76\u8de8\u4e91\u5382\u5546\u7684Kafka\u90e8\u7f72\u4f18\u5316<\/p>\n<\/li>\n
\u8fb9\u7f18\u8ba1\u7b97\u573a\u666f：\u5728\u8d44\u6e90\u53d7\u9650\u73af\u5883\u4e0b\u5b9e\u73b0\u9ad8\u6027\u80fd\u6d88\u606f\u4f20\u8f93<\/p>\n<\/li>\n
\u5b89\u5168\u4e0e\u6027\u80fd\u5e73\u8861：\u7814\u7a76\u52a0\u5bc6\u4f20\u8f93\u5bf9\u6027\u80fd\u7684\u5f71\u54cd\u548c\u4f18\u5316<\/p>\n<\/li>\n
\u901a\u8fc7\u672c\u6b21\u6df1\u5ea6\u5256\u6790，\u6211\u4eec\u89c1\u8bc1\u4e86Kafka\u5982\u4f55\u5c063\u53f0"\u5ec9\u4ef7"\u670d\u52a1\u5668\u7684\u6f5c\u529b\u53d1\u6325\u5230\u6781\u81f4，\u8fbe\u5230200\u4e07TPS\u7684\u60ca\u4eba\u6027\u80fd\u3002\u8fd9\u4e0d\u4ec5\u662f\u4e00\u6b21\u6280\u672f\u538b\u6d4b，\u66f4\u662f\u5bf9\u73b0\u4ee3\u5206\u5e03\u5f0f\u7cfb\u7edf\u8bbe\u8ba1\u54f2\u5b66\u7684\u4e00\u6b21\u6df1\u523b\u5c55\u793a\u3002<\/p>\n","protected":false},"excerpt":{"rendered":"
\u5f15\u8a00：\u6027\u80fd\u7684\u8fb9\u754c\u4e0e\u6210\u672c\u7684\u827a\u672f\u5728\u5206\u5e03\u5f0f\u7cfb\u7edf\u9886\u57df，\u6027\u80fd\u4e0e\u6210\u672c\u7684\u5e73\u8861\u4e00\u76f4\u662f\u4e2a\u6c38\u6052\u8bfe\u9898\u3002\u5f53\u4e1a\u754c\u666e\u904d\u8ba4\u4e3a\u5b9e\u73b0\u767e\u4e07\u7ea7TPS\u9700\u8981\u6602\u8d35\u7684\u9ad8\u7aef\u786c\u4ef6\u65f6，Kafka\u7528\u4e00\u7ec4\u4ee4\u4eba\u9707\u60ca\u7684\u6570\u636e\u6311\u6218\u4e86\u8fd9\u4e00\u8ba4\u77e5：\u4ec5\u4f7f\u75283\u53f0\\”\u5ec9\u4ef7\\”\u670d\u52a1\u5668，\u7adf\u80fd\u652f\u6491200\u4e07TPS\u3002\u8fd9\u4e00\u7ed3\u679c\u4e0d\u4ec5\u6253\u7834\u4e86\u786c\u4ef6\u51b3\u5b9a\u8bba\u7684\u8ff7\u601d，\u66f4\u63ed\u793a\u4e86\u73b0\u4ee3\u8f6f\u4ef6\u67b6\u6784\u4e0e\u4f18\u5316\u6280\u672f\u7684\u5de8\u5927\u6f5c\u529b\u3002\u672c\u6587\u5c06\u6df1\u5165\u5256\u6790\u8fd9\u4e00\u6027\u80fd\u5947\u8ff9\u80cc\u540e\u7684\u6280\u672f\u539f\u7406\u3001\u5b9e\u73b0\u8def\u5f84\u548c\u8c03\u4f18\u7ec6\u8282，\u4e3a\u60a8\u63d0\u4f9b\u4e00\u4efd\u5b8c<\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[55,3231],"topic":[],"class_list":["post-82858","post","type-post","status-publish","format-standard","hentry","category-server","tag-c","tag-linq"],"yoast_head":"\n