Windows系统下模拟Kafka集群参数调优

　　配置kafka运行环境
　　kafka的包中已经包含了zookeeper但是我还是喜欢单独下载一个zookeeper下载kafka
　　https：archive。apache。orgdistkafka2。6。3kafka2。122。6。3。tgz下载zookeeper
　　https：www。apache。orgdyncloser。luazookeeperzookeeper3。6。3apachezookeeper3。6。3bin。tar。gz配置zookeeper
　　解压下载的zookeeper安装包
　　复制三份并修改文件夹的名字（其实可以用三份配置但是我喜欢复制三份）修改点一新建zkData文件夹
　　在每一份的目录下新建zkData文件夹
　　修改点二新建myid文件
　　在每一份的zkData文件夹中新建myid文件，并分别把内容修改为123
　　修改点三修改配置文件
　　将conf目录下的zoosample。cfg文件修改为zoo。cfg，并修改文件内容
　　修改文件内容
　　第一处dataDir修改为zkData文件夹注意路径要用双反斜杠
　　第二处端口号因为是在一台机器上跑多个实例所以端口号不能冲突
　　第三处添加入下内容
　　server。1localhost：2881：3881server。2localhost：2882：3882server。3localhost：2883：3883
　　配置参数说明
　　server。AB：C：D
　　A是一个数字，表示这个是第几号服务器；
　　集群模式下配置一个文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo。cfg里面的配置信息比较从而判断到底是哪个server。
　　B是这个服务器的地址；
　　C是这个服务器Follower与集群中的Leader服务器交换信息的端口；
　　D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。
　　修改后的三个文件内容ThenumberofmillisecondsofeachticktickTime2000ThenumberofticksthattheinitialsynchronizationphasecantakeinitLimit10ThenumberofticksthatcanpassbetweensendingarequestandgettinganacknowledgementsyncLimit5thedirectorywherethesnapshotisstored。donotusetmpforstorage，tmphereisjustexamplesakes。dataDirD：softrepositoryzookeeperapachezookeeper3。6。3bin1zkDatatheportatwhichtheclientswillconnectclientPort2181themaximumnumberofclientconnections。increasethisifyouneedtohandlemoreclientsmaxClientCnxns60Besuretoreadthemaintenancesectionoftheadministratorguidebeforeturningonautopurge。http：zookeeper。apache。orgdoccurrentzookeeperAdmin。htmlscmaintenanceThenumberofsnapshotstoretainindataDirautopurge。snapRetainCount3PurgetaskintervalinhoursSetto0todisableautopurgefeatureautopurge。purgeInterval1MetricsProvidershttps：prometheus。ioMetricsExportermetricsProvider。classNameorg。apache。zookeeper。metrics。prometheus。PrometheusMetricsProvidermetricsProvider。httpPort7000metricsProvider。exportJvmInfotrueserver。1localhost：2881：3881server。2localhost：2882：3882server。3localhost：2883：3883启动zookeeper集群
　　在cmd中进入每一份的zookeeper文件夹中，执行binzkServer。cmd
　　配置kafka
　　在D盘根目录复制解压后的kafka三份修改configserver。properties
　　broker。id0
　　log。dirstmpkafkalogs1
　　zookeeper。connectlocalhost：2181，localhost：2182，localhost：2183kafka
　　port9091
　　端口号三份分别为909190929093LicensedtotheApacheSoftwareFoundation（ASF）underoneormorecontributorlicenseagreements。SeetheNOTICEfiledistributedwiththisworkforadditionalinformationregardingcopyrightownership。TheASFlicensesthisfiletoYouundertheApacheLicense，Version2。0（theLicense）；youmaynotusethisfileexceptincompliancewiththeLicense。YoumayobtainacopyoftheLicenseathttp：www。apache。orglicensesLICENSE2。0Unlessrequiredbyapplicablelaworagreedtoinwriting，softwaredistributedundertheLicenseisdistributedonanASISBASIS，WITHOUTWARRANTIESORCONDITIONSOFANYKIND，eitherexpressorimplied。SeetheLicenseforthespecificlanguagegoverningpermissionsandlimitationsundertheLicense。seekafka。server。KafkaConfigforadditionaldetailsanddefaultsServerBasicsTheidofthebroker。Thismustbesettoauniqueintegerforeachbroker。broker。id1SocketServerSettingsTheaddressthesocketserverlistenson。Itwillgetthevaluereturnedfromjava。net。InetAddress。getCanonicalHostName（）ifnotconfigured。FORMAT：listenerslistenername：hostname：portEXAMPLE：listenersPLAINTEXT：your。host。name：9092listenersPLAINTEXT：：9092Hostnameandportthebrokerwilladvertisetoproducersandconsumers。Ifnotset，itusesthevalueforlistenersifconfigured。Otherwise，itwillusethevaluereturnedfromjava。net。InetAddress。getCanonicalHostName（）。advertised。listenersPLAINTEXT：your。host。name：9092Mapslistenernamestosecurityprotocols，thedefaultisforthemtobethesame。Seetheconfigdocumentationformoredetailslistener。security。protocol。mapPLAINTEXT：PLAINTEXT，SSL：SSL，SASLPLAINTEXT：SASLPLAINTEXT，SASLSSL：SASLSSLThenumberofthreadsthattheserverusesforreceivingrequestsfromthenetworkandsendingresponsestothenetworknum。network。threads3Thenumberofthreadsthattheserverusesforprocessingrequests，whichmayincludediskIOnum。io。threads8Thesendbuffer（SOSNDBUF）usedbythesocketserversocket。send。buffer。bytes102400Thereceivebuffer（SORCVBUF）usedbythesocketserversocket。receive。buffer。bytes102400Themaximumsizeofarequestthatthesocketserverwillaccept（protectionagainstOOM）socket。request。max。bytes104857600LogBasicsAcommaseparatedlistofdirectoriesunderwhichtostorelogfileslog。dirstmpkafkalogs1Thedefaultnumberoflogpartitionspertopic。Morepartitionsallowgreaterparallelismforconsumption，butthiswillalsoresultinmorefilesacrossthebrokers。num。partitions1Thenumberofthreadsperdatadirectorytobeusedforlogrecoveryatstartupandflushingatshutdown。ThisvalueisrecommendedtobeincreasedforinstallationswithdatadirslocatedinRAIDarray。num。recovery。threads。per。data。dir1InternalTopicSettingsThereplicationfactorforthegroupmetadatainternaltopicsconsumeroffsetsandtransactionstateForanythingotherthandevelopmenttesting，avaluegreaterthan1isrecommendedtoensureavailabilitysuchas3。offsets。topic。replication。factor1transaction。state。log。replication。factor1transaction。state。log。min。isr1LogFlushPolicyMessagesareimmediatelywrittentothefilesystembutbydefaultweonlyfsync（）tosynctheOScachelazily。Thefollowingconfigurationscontroltheflushofdatatodisk。Thereareafewimportanttradeoffshere：1。Durability：Unflusheddatamaybelostifyouarenotusingreplication。2。Latency：Verylargeflushintervalsmayleadtolatencyspikeswhentheflushdoesoccurastherewillbealotofdatatoflush。3。Throughput：Theflushisgenerallythemostexpensiveoperation，andasmallflushintervalmayleadtoexcessiveseeks。ThesettingsbelowallowonetoconfiguretheflushpolicytoflushdataafteraperiodoftimeoreveryNmessages（orboth）。Thiscanbedonegloballyandoverriddenonapertopicbasis。Thenumberofmessagestoacceptbeforeforcingaflushofdatatodisklog。flush。interval。messages10000Themaximumamountoftimeamessagecansitinalogbeforeweforceaflushlog。flush。interval。ms1000LogRetentionPolicyThefollowingconfigurationscontrolthedisposaloflogsegments。Thepolicycanbesettodeletesegmentsafteraperiodoftime，orafteragivensizehasaccumulated。Asegmentwillbedeletedwhenevereitherofthesecriteriaaremet。Deletionalwayshappensfromtheendofthelog。Theminimumageofalogfiletobeeligiblefordeletionduetoagelog。retention。hours168Asizebasedretentionpolicyforlogs。Segmentsareprunedfromthelogunlesstheremainingsegmentsdropbelowlog。retention。bytes。Functionsindependentlyoflog。retention。hours。log。retention。bytes1073741824Themaximumsizeofalogsegmentfile。Whenthissizeisreachedanewlogsegmentwillbecreated。log。segment。bytes1073741824Theintervalatwhichlogsegmentsarecheckedtoseeiftheycanbedeletedaccordingtotheretentionpolicieslog。retention。check。interval。ms300000ZookeeperZookeeperconnectionstring（seezookeeperdocsfordetails）。Thisisacommaseparatedhost：portpairs，eachcorrespondingtoazkserver。e。g。127。0。0。1：3000，127。0。0。1：3001，127。0。0。1：3002。Youcanalsoappendanoptionalchrootstringtotheurlstospecifytherootdirectoryforallkafkaznodes。zookeeper。connectlocalhost：2181，localhost：2182，localhost：2183kafkaTimeoutinmsforconnectingtozookeeperzookeeper。connection。timeout。ms18000GroupCoordinatorSettingsThefollowingconfigurationspecifiesthetime，inmilliseconds，thattheGroupCoordinatorwilldelaytheinitialconsumerrebalance。Therebalancewillbefurtherdelayedbythevalueofgroup。initial。rebalance。delay。msasnewmembersjointhegroup，uptoamaximumofmax。poll。interval。ms。Thedefaultvalueforthisis3seconds。Weoverridethisto0hereasitmakesforabetteroutoftheboxexperiencefordevelopmentandtesting。However，inproductionenvironmentsthedefaultvalueof3secondsismoresuitableasthiswillhelptoavoidunnecessary，andpotentiallyexpensive，rebalancesduringapplicationstartup。group。initial。rebalance。delay。ms0port9091启动kafkabinwindowskafkaserverstart。batconfigserver。propertiestopic查看topicbinwindowskafkatopics。batbootstrapserverlocalhost：9091list创建topicbinwindowskafkatopics。batbootstrapserverlocalhost：9091createpartitions1replicationfactor1topichellotest查看topic的详情binwindowskafkatopics。batbootstrapserverlocalhost：9091describetopichellotest修改分区数
　　注意：分区数只能增加，不能减少binwindowskafkatopics。batbootstrapserverlocalhost：9091altertopichellotestpartitions3删除topicbinwindowskafkatopics。batbootstrapserverlocalhost：9091deletetopichello集群压力测试
　　用Kafka自带的脚本，对Kafka进行压测。
　　生产者压测脚本：kafkaproducerperftest。sh
　　消费者压测脚本：kafkaconsumerperftest。shKafkaProducer压力测试创建一个hellotesttopic，设置为3个分区3个副本binwindowskafkatopics。batbootstrapserverlocalhost：9091createreplicationfactor3partitions3topichellotest测试生产者binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput1producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size16384linger。ms0
　　recordsize是一条信息有多大，单位是字节，本次测试设置为1k。
　　numrecords是总共发送多少条信息，本次测试设置为100万条。
　　throughput是每秒多少条信息，设成1，表示不限流，尽可能快的生产数据，可测出生产者最大吞吐量。
　　producerprops后面可以配置生产者相关参数，batch。size配置为16k调整batch。size大小
　　batch。size默认值是16k将batch。size设置为32kbinwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size32768linger。ms0
　　batch。size默认值是16k。将batch。size设置为4k。binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms0调整linger。ms
　　linger。ms默认是0ms。将linger。ms设置为50ms。binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms50压缩方式
　　调整压缩方式默认的压缩方式是none。将compression。type设置为snappy。binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms50compression。typesnappy
　　默认的压缩方式是none。将compression。typee设置为zstd。binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms50compression。typezstd
　　默认的压缩方式是none。将compression。type设置为gzip。binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms50compression。typegzip
　　默认的压缩方式是none。将compression。type设置为lz4binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput10000producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms50compression。typelz4调整缓存大小buffer。memory
　　调整缓存大小默认生产者端缓存大小32m。将buffer。memory设置为64m。binwindowskafkaproducerperftest。battopichellotestrecordsize1024numrecords1000000throughput1producerpropsbootstrap。serverslocalhost：9091，localhost：9092，localhost：9093batch。size4096linger。ms50buffer。memory67108864测试Consumer性能默认情况的吞吐量binwindowskafkaconsumerperftest。batbootstrapserverlocalhost：9091，localhost：9092，localhost：9093topichellotestmessages10000000consumer。configconfigconsumer。propertiestimeout10000
　　bootstrapserver指定Kafka集群地址
　　topic指定topic的名称
　　messages总共要消费的消息个数
　　调整max。poll。records
　　一次拉取条数为2000
　　修改configconsumer。properties文件中的一次拉取条数为2000max。poll。records2000
　　一次拉取条数为20000
　　修改configconsumer。properties文件中的一次拉取条数为20000max。poll。records20000
　　一次拉取条数为200000
　　修改configconsumer。properties文件中的一次拉取条数为200000max。poll。records200000
　　可以看到每秒的消息数量已经不会提升了
　　一次拉取条数为100000
　　修改configconsumer。properties文件中的一次拉取条数为100000max。poll。records100000
　　fetch。max。bytes
　　修改optmodulekafkaconfigconsumer。properties文件中的拉取一批数据大小100mfetch。max。bytes104857600