Брокерски конфигурации
| Имоти | По подразбиране | Описание | | broker.id | | Всеки брокер може да бъде идентифициран с уникален неотрицателен целочислен идентификатор; Този id може да се използва като "име" на брокера, а съществуването му позволява на брокера да мигрира към друг хост/порт, без да обърква потребителите. Можеш да избереш каквото и да е число като ID, стига ID-то да е уникално. | | log.dirs | /tmp/kafka-logs | Пътят, по който Кафка съхранява данни. Този път не е уникален, може да бъде множествен, и пътищата трябва да се разделят само с запетаи; Винаги когато се създава нов дял, той се избира да го направи по пътя, който съдържа най-малък брой дялове. | | Пристанище | 6667 | Сървърът приема порта, към който клиентът се свързва | | zookeeper.connect | null | Форматът на свързващия низ на ZooKeeper е: hostname:port, където hostname и port са съответно хостът и портът на възел в клъстера ZooKeeper. За да се свържете с други ZooKeeper възли, когато хостът спре, можете да създадете няколко хоста по следния начин:
hostname1:port1, hostname2:port2, hostname3:port3.
ZooKeeper ви позволява да добавите "chroot" път, за да съхранявате всички kafka данни в клъстера под конкретен път. Когато няколко Kafka клъстера или други приложения използват един и същ ZooKeeper клъстер, можете да използвате този метод за задаване на пътя за съхранение на данни. Това може да се реализира чрез форматиране на свързващия низ по следния начин:
Име на хост1:порт1,име на хост2:порт2,име на хост3:порт3/chroot/path
Тази конфигурация съхранява всички данни от Kafka клъстера под пътя /chroot/path. Имайте предвид, че преди да стартирате брокера, трябва да създадете този път, а потребителите трябва да използват същия формат на връзка. | | message.max.bytes | 1000000 | Максималният размер на съобщенията, които сървърът може да получи. Важно е настройките на потребителя и производителя относно това свойство да бъдат синхронизирани, в противен случай съобщението, публикувано от производителя, е твърде голямо за потребителя. | | num.network.threads | 3 | Броят на мрежовите нишки, използвани от сървъра за обработка на мрежови заявки; Обикновено не е нужно да променяте този имот. | | num.io.threads | 8 | Броят на входно-изходните нишки, използвани от сървъра за обработка на заявки; Броят на нишките трябва да е поне равен на броя твърди дискове. | | background.threads | 4 | Броят на нишките, използвани за фонова обработка, като изтриване на файлове; Не е нужно да променяте този имот. | | queued.max.requests | 500 | Максималният брой заявки, които могат да бъдат поставени на опашка за обработка на I/O нишка, преди мрежовата нишка да спре да чете нови заявки. | | host.name | null | хост на брокера; Ако името на хоста вече е зададено, брокерът ще бъде обвързан само с този адрес; Ако няма настройка, ще се свърже към всички интерфейси и ще публикува едно копие на ZK | | advertised.host.name | null | Ако е зададено, то се изпраща на производители, потребители и други брокери като хост име на брокера | | ad.port | null | Това пристанище се предоставя на производители, потребители и други брокери и се използва за установяване на връзка; Трябва да се настрои само ако самият порт и портът, който сървърът трябва да свърже, са различни. | | socket.send.buffer.bytes | 100 * 1024 | SO_SNDBUFF Размер на кеша, който сървърът използва за връзки между сокети | | socket.receive.buffer.bytes | 100 * 1024 | SO_RCVBUFF размера на кеша, който се използва от сървъра за свързване към сокети | | socket.request.max.bytes | 100 * 1024 * 1024 | Максималният размер на заявката, разрешен от сървъра; Това ще избегне препълване на сървъра, което би трябвало да е по-малко от размера на купчината в Java | | num.partitions | 1 | Ако броят на дяловете не е даден при създаване на тема, това число ще бъде стандартният брой дялове под темата. | | log.segment.bytes | 1014*1024*1024 | Логовете на темата се съхраняват в много файлове в определена директория, които разделят логовете на дяла на сегменти; Този атрибут е максималният размер на всеки файл; Когато размерите достигнат тази стойност, се създава нов файл. Тази настройка може да бъде заменена от основата на всяка тема.
Изглед Входът към хиперлинк е видим. | | log.roll.hours | 24 * 7 | Дори ако файлът не достигне log.segment.bytes, нов файл се създава всеки път, когато времето за създаване достигне това свойство. Тази настройка може да бъде заместена и от настройки на тематично ниво;
ИзгледВходът към хиперлинк е видим. | | log.cleanup.policy | Изтрий | | | log.retention.minutes и log.retention.hours | 7 дни | Времето, в което всеки лог файл е бил запазен преди да бъде изтрит. Стандартното време за пестене на данни е едно и също за всички теми.
log.retention.minutes и log.retention.bytes се използват за задаване на изтриването на log файлове, независимо кое свойство е препълнено.
Тази настройка на свойството може да бъде преразгледана, когато темата е практически зададена по принцип.
ИзгледВходът към хиперлинк е видим. | | log.retention.bytes | -1 | Общото количество данни, запазени от всеки дял под всяка тема; Забележете, че това е горната граница за всяка част, така че това число, умножено по броя на дяловете, е общото количество данни, съхранени на тема. Също така: Ако са зададени и log.retention.hours, и log.retention.bytes, превишаването на някое от ограниченията ще доведе до изтриване на сегментен файл.
Имайте предвид, че тази настройка може да бъде заменена от всяка тема.
ИзгледВходът към хиперлинк е видим. | | log.retention.check.interval.ms | 5 минути | Проверете интервала между лог-сегментираните файлове, за да определите дали атрибутите отговарят на изискванията за изтриване. | | log.cleaner.enable | false | Когато това свойство е зададено на false, то ще бъде изтрито, след като дневникът бъде съхранен за максимално време или размер. Ако е настроено на true, това ще се случи, когато атрибутът за запазване достигне горната границаВходът към хиперлинк е видим.。 | | log.cleaner.threads | 1 | Броят на нишките, извършващи компресия на логаритми | | log.cleaner.io.max.bytes.per.sec | Никой | Максималният брой I/O операции, които почиствачът на логове може да има при извършване на уплътняване на логове. Тази настройка ограничава почистващия да избегне намеса в активните услуги за заявки. | | log.cleaner.io.buffer.size | 500*1024*1024 | Log Cleaner индексира логовете по време на процеса на почистване и намалява размера на използвания кеш. Най-добре е да го настроите голям, за да осигурите достатъчно памет. | | log.cleaner.io.buffer.load.factor | 512*1024 | Размерът на входно-изходния блок, необходим за почистване на трупи. Не е нужно да променяш тази настройка. | | log.cleaner.io.buffer.load.factor | 0.9 | коефициент на натоварване на хеш таблицата, използвана при почистване на логове; Не е нужно да променяте тази опция. | | log.cleaner.backoff.ms | 15000 | Периодът, в който трупът се почиства, се извършва | | log.cleaner.min.cleanable.ratio | 0.5 | Тази конфигурация контролира колко често компакторът на трупите се опитва да почисти трупите (предполага се, чеВходът към хиперлинк е видим.е отворена). По подразбиране избягвайте почистването на повече от 50% от трупите. Това съотношение е свързано с максималното пространство, заето от архивния лог (50% от логовете се компресират при 50%). По-високата ставка означава по-малко отпадъци и повече пространство, което може да се изчисти по-ефективно. Тази настройка може да бъде преразгледана във всяка тематична настройка.
ИзгледВходът към хиперлинк е видим.。 | | log.cleaner.delete.retention.ms | 1 ден | време за съхранение; Максималното време за съхранение на компресирани логове; Това е и максималното време, за което клиентът консумира съобщения, а разликата между log.retention.minutes е, че единият контролира некомпресираните данни, а другият – компресираните, които ще бъдат презаписани с посоченото време, когато темата е създадена. | | log.index.size.max.bytes | 10*1024*1024 | Максималният размер на сегмент от дърво. Имайте предвид, че ако размерът на log достигне тази стойност, трябва да се генерира нов log сегмент, дори ако размерът не надвишава лимита log.segment.bytes. | | log.index.interval.bytes | 4096 | При извършване на fetch трябва да сканирате най-близкия офсет с определено количество пространство, колкото по-голяма е настройката, толкова по-добре, обикновено използвайте стандартната стойност | | log.flush.interval.messages | Long.MaxValue | лог файлът "синхронизира" се с диска преди натрупване на съобщения. Тъй като операциите с дисков вход са бавна операция, но също така необходим начин за "надеждност на данните", проверете дали е необходим времеви интервал за втвърдяване на твърдия диск. Ако тази стойност е твърде голяма, това ще доведе до твърде дълго време за "синхронизация" (блокиране на вход), ако е твърде малка, ще доведе до дълго време на "fsync" (блокиране на входа), ако тази стойност е твърде малка, ще доведе до голям брой "синхронизирани" моменти, което означава, че цялостната заявка на клиента има определено забавяне, а повредата на физическия сървър ще доведе до загуба на съобщения без fsync. | | log.flush.scheduler.interval.ms | Long.MaxValue | Проверете дали са необходими fsync интервали | | log.flush.interval.ms | Long.MaxValue | Това число се използва за контрол на времевия интервал на "fsync", ако броят на съобщенията никога не достигне броя на съобщенията, фиксирани на диска, но интервалът от последната синхронизация на диска достигне прага, синхронизацията на диска също се задейства. | | log.delete.delay.ms | 60000 | Времето за съхранение след изчистване на файла в индекса обикновено не е необходимо да се променя | | auto.create.topics.enable | true | Дали да се позволи автоматично създаване на теми. Ако е вярно, автоматично ще създаде тема, която не съществува, когато produce или fetch не съществуват. В противен случай трябва да използвате командния ред, за да създадете тема | | controller.socket.timeout.ms | 30000 | Времето за тайм-аут на сокета, когато контролерът за управление на дялове извършва резервно копие. | | controller.message.queue.size | Int.MaxValue | Контролер към брокер-чанъл | | default.replication.factor | 1 | По подразбиране броят на резервните копия се отнася само до автоматично създадени теми | | replica.lag.time.max.ms | 10000 | Ако последователят не изпрати заявка за хвърляне в този период, лидерът ще премахне последователя от ISR и ще го счита за обесен | | replica.lag.max.съобщения | 4000 | Ако репликата има повече от този брой неподкрепени, лидерът премахва последователя и го счита за обесен | | replica.socket.timeout.ms | 30*1000 | Време за тайм-аут на лидера за заявки от Socket Network при архивиране на данни | | replica.socket.receive.buffer.bytes | 64*1024 | Socket приемащ буфер при изпращане на мрежова заявка към лидера по време на архивиране | | replica.fetch.max.bytes | 1024*1024 | Максималната стойност на всяко изтегляне към момента на архивиране | | replica.fetch.min.bytes | 500 | Максималното време за изчакване, докато данните достигнат до лидера, когато лидерът направи заявка за резервно копие | | replica.fetch.min.bytes | 1 | Най-малкият размер на отговора след всяко изтегляне при архивиране | | num.replica.fetchers | 1 | Броят на нишките, които архивират данни от лидера | | replica.high.watermark.checkpoint.interval.ms | 5000 | Всяка реплика проверява колко често се втвърдява най-високото ниво на водата | | fetch.purgatory.purge.interval.requests | 1000 | Fetch заявка за purge интервала | | producer.purgatory.purge.interval.requests | 1000 | Производителят иска интервал за почистване | | zookeeper.session.timeout.ms | 6000 | Тайм-аут на сесията на пазача на зоопарка. | | zookeeper.connection.timeout.ms | 6000 | Максималното време, което клиентът чака, за да установи връзка със зоопазача | | zookeeper.sync.time.ms | 2000 | ZK Follower изостава от лидера на ZK за най-дълго време | | controlled.shutdown.enable | true | Дали е възможно да се контролира затварянето на брокера. Ако е възможно, брокерът ще може да премести всички лидери към други брокери преди финализирането на сделката. Това намалява недостъпността по време на процеса на изключване. | | controlled.shutdown.max.опитва отново | 3 | Броят на командите, които могат успешно да изпълнят изключване преди непълно изключване. | | controlled.shutdown.retry.backoff.ms | 5000 | Време за откъсване между затварянията | | auto.leader.rebalance.enable | true | Ако това е вярно, контролерът автоматично ще балансира лидерството на брокерите по дяловете | | leader.imbalance.per.broker.percent | 10 | Максималният коефициент на дисбаланс, разрешен от всеки брокер | | leader.imbalance.check.interval.seconds | 300 | Проверете честотата на дисбаланса на лидера | | offset.metadata.max.bytes | 4096 | Позволява на клиентите да запазят максималния брой офсети | | max.connections.per.ip | Int.MaxValue | Максималният брой връзки на брокер може да се осъществи към всеки IP адрес | | max.connections.per.ip.overrides | | Максималното покритие на стандартната връзка на IP адрес или хост име | | connections.max.idle.ms | 600000 | Лимит на тайм-аут за празни връзки | | log.roll.jitter. {MS,HOURS} | 0 | Максималният брой треперения, извлечени от logRollTimeMillis | | num.recovery.threads.per.data.dir | 1 | Броят нишки, които всяка директория с данни използва за записване на възстановяването | | нечисто.лидер.избор.активирай | true | Показва дали е възможно да се използва настройката без реплики в ISR като лидер | | delete.topic.enable | false | Възможност за изтриване на теми | | offsets.topic.num.partitions | 50 | Броят на дяловете за темата за офсет коммит. Тъй като промяната на това след внедряване в момента не се поддържа, препоръчваме да се използва по-висока настройка за продукция (например 100-200). | | offsets.topic.retention.minutes | 1440 | Офсети, които са съществували по-дълго от този срок, ще бъдат маркирани като чакащи изтривания | | offsets.retention.check.interval.ms | 600000 | Мениджърът на офсетите проверява честотата на застояли отклонения | | offsets.topic.replication.factor | 3 | Броят на резервните копия на темата е изместен. Препоръчва се да се задават по-високи стойности, за да се гарантира по-голяма наличност | | offset.topic.segment.bytes | 104857600 | Темата се измества. | | offsets.load.buffer.size | 5242880 | Тази настройка е свързана с размера на партидата и се използва при четене от сегмента с офсети. | | offsets.commit.required.acks | -1 | Броят на потвържденията трябва да се зададе, преди офсетният коммит да стане приемлив, и обикновено не се налага да се променя |
| Имоти | По подразбиране | Стандартно свойство на сървъра | Описание | | cleanup.policy | Изтрий | log.cleanup.policy | Или "изтрий", или "компактен"; Този низ показва как да се използва старата логарифмична част; Стандартният метод ("delete") изхвърля старата част, когато времето за рециклиране или лимита за размер бъде достигнат. "компактен" ще компресира логовете | | delete.retention.ms | 86400000 (24 часа) | log.cleaner.delete.retention.ms | Разликата между log.retention.minutes е, че едната управлява некомпресирани данни, а другата – компресирани данни. Тази конфигурация може да бъде презаписана от параметрите за закрепване, когато темата е създадена | | Flush.съобщения | Никой | log.flush.interval.messages | Тази конфигурация определя времеви интервал за принудително използване на fsync логове. Например, ако тази опция е зададена на 1, тогава fsync е необходим след всяко съобщение, а ако е зададено на 5, fsync е необходим за всеки 5 съобщения. Обикновено се препоръчва да не задавате тази стойност. Настройката на този параметър изисква необходимия компромис между "надеждност на данните" и "производителност". Ако тази стойност е твърде голяма, това ще доведе до дълго време за "fsync" всеки път (блокиране на входа), а ако тази стойност е твърде малка, ще доведе до голям брой "fsync", което също означава, че има известно забавяне в цялостната заявка на клиента. Повреда на физическия сървър ще доведе до загуба на съобщения без fsync. | | flush.ms | Никой | log.flush.interval.ms | Тази конфигурация се използва за фиксиране на времевия интервал между принуждаването на fsync логове към диска; Например, ако е настроено на 1000, тогава се изисква fsync на всеки 1000ms. Тази опция обикновено не се препоръчва | | index.interval.bytes | 4096 | log.index.interval.bytes | Настройката по подразбиране гарантира, че добавяме индекс към съобщението на всеки 4096 байта, а повече индекси правят съобщението за четене по-близо, но размерът на индекса ще бъде увеличен. Тази опция обикновено не е задължителна | | max.message.bytes | 1000000 | max.message.bytes | Максималният размер на съобщението за добавяне на кафка. Имайте предвид, че ако увеличите този размер, трябва да увеличите и размера на fetch на потребителя, за да може потребителят да извлича съобщения до тези максимални размери. | | минимално.почистващо.мръсно.съотношение | 0.5 | минимално.почистващо.мръсно.съотношение | Тази конфигурация контролира колко често компресорът на трупи се опитва да изчисти трупите. По подразбиране логове с компресия над 50% ще се избягват. Това съотношение избягва най-голямата загуба на пространство | | min.insync.replicas | 1 | min.insync.replicas | Когато producer е настроен на request.required.acks на -1, min.insync.replicas посочва минималния брой реплики (всяко repica записване трябва да е успешно), и ако този брой не бъде достигнат, производителят ще създаде изключение. | | retention.bytes | Никой | log.retention.bytes | Ако използвате политиката за задържане "delete", тази конфигурация се отнася до максималния размер, който логът може да достигне преди да бъде изтрит. По подразбиране няма ограничение на размера, а само времево ограничение | | retention.ms | 7 дни | log.retention.minutes | Ако използвате политиката за задържане "delete", тази конфигурация се отнася до времето, когато логът е бил запазен преди лога за изтриване. | | segment.bytes | 1GB | log.segment.bytes | В Kafka логаритмичните записи се съхраняват на парчета, като тази конфигурация се отнася до размера на логаритмичните логаритми, разделени на части | | segment.index.bytes | 10MB | log.index.size.max.bytes | Тази конфигурация е приблизително с размера на индексния файл, съпоставен между офсетите и локациите на файловете; Тази конфигурация обикновено не се нуждае от промяна | | segment.ms | 7 дни | log.roll.hours | Дори ако log chunk файлът не достигне размера, който трябва да бъде изтрит или компресиран, след като времето за log достигне този горен лимит, ще бъде принудително създаден нов log chunk файл | | segment.jitter.ms | 0 | log.roll.jitter. {MS,HOURS} | Максималният джитър да се извади от logRollTimeMillis. |
Потребителски конфигурации
| Имоти | По подразбиране | Описание | | group.id | | Низ, който уникално идентифицира групата, в която се намира потребителският процес, и ако е зададен един и същ идентификатор на групата, това означава, че тези процеси принадлежат на една и съща потребителска група | | zookeeper.connect | | Посочете низа на връзката Zookeeper, форматът е hostname:port, тук хостът и портът са както хостът, така и портът на сървъра на Zookeeper, за да избегнете загуба на контакт след повреда на машината на Zookeeper, можете да зададете няколко hostname:ports, използвайки запетаи като разделяне:
Име на хоста1:порт1,име на хост2:порт2,име на хост3:порт3
Можете да добавите пътя на chroot на ZooKeeper към низа за връзка на ZooKeeper, който се използва за съхранение на собствените му данни, по следния начин:
Име на хост1:порт1,име на хост2:порт2,име на хост3:порт3/chroot/path | | consumer.id | null | Не е необходима настройка и обикновено се генерира автоматично | | socket.timeout.ms | 30*100 | Лимити за тайм-аут за мрежови заявки. Истинският лимит за тайм-аут е max.fetch.wait+socket.timeout.ms | | socket.receive.buffer.bytes | 64*1024 | Socket се използва за получаване на кеша размера на мрежовите заявки | | fetch.message.max.bytes | 1024*1024 | Максималният брой байтове на съобщение за извличане на заявка за извличане. Тези байтове ще бъдат наблюдавани в паметта, използвана за всеки дял, така че тази настройка ще контролира количеството памет, използвана от потребителя. Размерът на заявката за извличане трябва да е поне равен на максималния размер на съобщението, разрешен от сървъра, в противен случай размерът на съобщението, което производителят може да изпрати, е по-голям от размера, който потребителят може да консумира. | | num.consumer.fetchers | 1 | Броят на фичер нишките, използвани за извличане на данни | | auto.commit.enable | true | Ако е вярно, отместването на съобщението, изтеглено от потребителя, автоматично ще бъде синхронизирано със зоопазача. Този commit offset ще бъде използван от новия потребител, когато процесът приключи | | auto.commit.interval.ms | 60*1000 | Честотата, с която потребителят подава офсет на гледача в зоопарка, е в секунди | | queued.max.message.chunks | 2 | Максималният брой съобщения, използвани за кеширане за консумация. Всеки chunk трябва да е същият като fetch.message.max.bytes | | rebalance.max.опитва отново | 4 | Когато нов потребител се добави към потребителска група, колекцията от потребители се опитва да ребалансира броя на разпределените дялове, разпределени на всеки потребител. Ако колекцията на потребителите се промени, това ребалансиране се проваля и се реентрификува при изпълнението на разпределението | | fetch.min.bytes | 1 | Минималният брой байтове, които сървърът трябва да върне при всяка заявка за извличане. Ако не се върнат достатъчно данни, заявката изчаква, докато бъдат върнати достатъчно данни. | | fetch.wait.max.ms | 100 | Ако няма достатъчно данни за fetch.min.bytes, тази конфигурация се отнася до максималното време, което сървърът ще блокира преди да отговори на заявка за извличане. | | rebalance.backoff.ms | 2000 | Време е за отдръпване преди да опиташ отново ребланс | | refresh.leader.backoff.ms | 200 | Има време за изчакване, преди да се определи дали лидерът на разделението е загубил лидерството си | | auto.offset.reset | Най-голям | Ако няма инициализиран офсет в zookeeper, ако offset е отговор на следната стойност:
най-малко: автоматично нулиране на отместване към най-малкото отклонение
най-голямият: Автоматично нулиране на отместване към отклонението на най-голямото
Всичко друго: Прави изключение за потребителя | | consumer.timeout.ms | -1 | Ако няма налично съобщение, дори след изчакване за определено време, се хвърля изключение за тайм-аут | | exclude.internal.topics | true | Дали да се разкриват послания от вътрешни теми пред потребителите | | paritition.assignment.strategy | Ареал | Изберете политиката за прислушване на дялове към потребителския поток, по желание диапазон, кръгова робина | | client.id | Стойност на идентификатора на групата | е специфичен за потребителя низ, който помага за проследяване на повиквания във всяка заявка. Логично трябва да потвърди приложението, което е генерирало заявката | | zookeeper.session.timeout.ms | 6000 | Ограничения за тайм-аут за сесии с гледач на зоопарка. Ако потребителят не изпрати съобщение за сърдечен ритъм на зоопарка през това време, то се счита за блокирано и ще се генерира ребланс | | zookeeper.connection.timeout.ms | 6000 | Максималното време за изчакване за клиент да установи връзка със Zookeeper | | zookeeper.sync.time.ms | 2000 | Последователите на ZK могат да изостават лидера на ZK за максимално време | | offsets.storage | Пазач на зоопарк | Места, използвани за съхранение на офсети: zookeeper или kafka | | offset.channel.backoff.ms | 1000 | Времето за откъсване при повторно свързване към канала за отместване или повторен опит за извличане/потвърждение на неуспешния офсет | | offsets.channel.socket.timeout.ms | 10000 | Лимитът за тайм-аут на сокета за отговора на заявката за извличане/потвърждаване при четене на офсета. Този лимит на тайм-аут се използва от заявката consumerMetadata за заявка за управление на офсет | | offsets.commit.max.опитва отново | 5 | Броят пъти, в които е бил повторен офсет коммитът. Този опит се прилага само за офсет комити между изключването. него | | dual.commit.enabled | true | Ако използвате "kafka" като offsets.storage, можете да комитирате offset към zookeeper два пъти (и веднъж на kafka). Това е задължително при миграция от офсет съхранение базирано на зоопазител към кафка-базирано на офсет. За всяка потребителска група е безопасна препоръка да се изключи тази опция, когато миграцията приключи | | partition.assignment.strategy | Ареал | Изберете между политиките "диапазон" и "roundrobin" като политика за прислушване на дялове към потребителските потоци от данни; Циркулярният дялове разпределя всички налични дялове, както и всички налични потребителски нишки. Той ще присвои цикъла на дяловете на потребителската нишка. Ако всички потребителски екземпляри са абонирани към определен, дяловете се разделят на детерминистични разпределения. Стратегията за разпределение по кръгов систем е възможна само ако са изпълнени следните условия: (1) Всяка тема има еднакъв брой потоци от данни според силата на потребителя. (2) Колекцията от абонирани теми се определя за всеки потребителски инстанция в потребителската група. |
Конфигурации на продуцентите
| Имоти | По подразбиране | Описание | | metadata.broker.list | | Сервирай bootstrapping. Producer се използва само за получаване на метаданни (теми, дятове, реплики). Връзката с сокета за изпращане на действителните данни ще бъде установена въз основа на върнатите метаданни. Форматът е:
хост1:порт1,хост2:порт2
Този списък може да бъде подсписък с брокери или VIP, сочещ към брокери | | request.required.acks | 0 | Тази конфигурация е потвърждение, което показва кога заявката за продукция се счита за завършена. По-специално, колко други брокери трябва да са подали данни в логовете си и са потвърдили тази информация пред лидера си. Типичните стойности включват:
0: Показва, че производителят никога не чака потвърждение от брокера (същото поведение като 0.7). Тази опция осигурява най-малка латентност, но в същото време и най-голям риск (защото данните се губят, когато сървърът не работи).
1: Показва, че репликата на лидера е получила потвърждение за данните. Тази опция има ниска латентност и гарантира, че сървърът потвърждава, че е получена.
-1: Производителят получава потвърждение, че всички синхронизирани реплики са получили данни. Латентността е най-голяма, но този метод не елиминира напълно риска от загуба на съобщения, тъй като броят на синхронизираните реплики може да бъде 1. Ако искате да сте сигурни, че някои реплики получават данни, трябва да зададете опцията min.insync.replicas в настройките на тема. Прочетете дизайнерската документация за по-задълбочена дискусия. | | request.timeout.ms | 10000 | Брокерът прави всичко възможно да реализира изискването request.required.acks, в противен случай ще бъде изпратена грешка към клиента | | producer.type | синхронизация | Тази опция определя дали съобщението е изпратено асинхронно във фонова нишка. Правилни стойности:
(1) асинхронно: Асинхронно изпращане
(2) sync: Синхронизирано изпращане
Като зададем производителя на асинхронен режим, можем да обработваме заявки на партиди (което е добре за по-висока пропускателна способност), но това също така създава възможност клиентската машина да загуби неизпратени данни | | serializer.class | kafka.serializer.DefaultEncoder | Категорията сериализация на съобщението. Стандартният енкодер въвежда един байт[] и връща същия байт[] | | key.serializer.class | | Клас по сериализация за ключови думи. Ако това не е дадено, по подразбиране е да се съвпадне съобщението | | partitioner.class | kafka.producer.DefaultPartitioner | клас за разделяне на съобщения между подтеми. Стандартният разделител се базира на хеш таблицата на ключа | | compression.codec | Никой | Този параметър може да зададе кодека за компресиране на данни, който може да бъде избран като "none", "gzip", "snappy". | | compressed.topics | null | Този параметър може да се използва за определяне дали определени теми са компресирани. Ако компресираният кодек е кодек, различен от NoCompressCodec, тези кодеци се прилагат към зададените теми данни. Ако списъкът с компресирани теми е празен, приложи конкретния компресиран кодек към всички теми. Ако компресираният кодек е NoCompressionCodec, компресията не е достъпна за всички теми. | | message.send.max.опитва отново | 3 | Този параметър ще накара производителя автоматично да опитва неуспешни заявки за изпращане. Този параметър определя броя на повторенията. Забележка: Задаването на стойност извън 0 ще доведе до повторение на определени мрежови грешки: изпращане и загуба на потвърждение | | retry.backoff.ms | 100 | Преди всяко повторно опитване, продуцентът актуализира метаданните на съответната тема, за да види дали новият лидер е назначен. Тъй като изборът на лидер отнема малко време, тази опция определя колко време производителят трябва да изчака преди да актуализира метаданните. | | topic.metadata.refresh.interval.ms | 600*1000 | Продуцентът обикновено актуализира метаданните на темата при някои случаи на отказ (липсва дял, лидер е недостъпен и др.). Той ще премине през редовен цикъл. Ако го зададеш на отрицателна стойност, метаданните ще се обновят само ако се провалят. Ако е зададено на 0, метаданните се обновяват след всяко изпращане на съобщение (тази опция не се препоръчва, тъй като системата консумира твърде много). Важно: Актуализациите се случват само след като съобщението бъде изпратено, така че ако производителят никога не изпрати съобщението, метаданните никога не се обновяват. | | queue.buffering.max.ms | 5000 | Максималният времеви интервал, в който потребителят кешира данните, когато се прилага асинхронен режим. Например, ако съобщението е настроено на 100, съобщенията в рамките на 100ms ще се обработват на партиди. Това ще подобри пропускателната способност, но ще увеличи латентността поради кеширането. | | queue.buffering.max.съобщения | 10000 | При използване на асинхронен режим максималният брой неизпратени съобщения, които могат да бъдат кеширани в опашката преди производителя, трябва да бъде блокиран, иначе данните трябва да бъдат загубени | | batch.num.messages | 200 | При използване на асинхронен режим можете да обработвате пакетно максималния брой съобщения. Или броят на съобщенията е стигнал онлайн или queue.buffer.max.ms е пристигнал, и продуцентът ще ги обработи | | send.buffer.bytes | 100*1024 | Размер на кеша за записване на сокет | | client.id | “” | Този клиентски идентификатор е специфичен за потребителя низ, който е включен във всяка заявка за проследяване на обаждането, и логично би трябвало да може да потвърди, че приложението е направило заявката. |
Конфигурации на продуцентите
| Име | Тип | По подразбиране | Значение | Описание | | boostrap.servers | Списък | | високо | Хост/порт група за установяване на връзка с клъстера Kafka. Данните ще се зареждат равномерно на всички сървъри, независимо кой сървър е предназначен за стартиране. Този списък засяга само инициализираните хостове (които се използват за откриване на всички сървъри). Този формат на списъка:
host1:port1,host2:port2,...
Тъй като тези сървъри се използват само за инициализация на връзки, за да се открият всички членове на клъстера (които могат да се променят динамично), този списък не е задължително да съдържа всички сървъри (може да искате повече от един сървър, въпреки че в този случай един сървър може да не работи). Ако няма сървър в този списък, изпращането на данни ще се провали, докато списъкът не стане наличен. | | ACKS | Низ | 1 | високо | Производителят се нуждае от сигнал от сървъра, за да потвърди получаването след получаване на данните, като тази конфигурация се отнася до броя на такива сигнали за потвърждение, от които прокудерът се нуждае. Тази конфигурация всъщност представлява наличието на резервни копия на данни. Следните настройки са често срещани опции:
(1) acks=0: Зададено на 0 означава, че производителят не трябва да чака потвърждение на получената информация. Репликата веднага ще бъде добавена в буфера на сокет и ще се счита за изпратена. В този случай няма гаранция, че сървърът е получил успешно данните, а повторният опит на конфигурацията няма да работи (защото клиентът не знае дали е неуспешен) и отклонението на обратната връзка винаги ще бъде зададено на -1.
(2) acks=1: Това означава, че поне трябва да изчакате лидерът успешно да запише данните в локалния лог, но не всички последователи да напишат успешно. В този случай, ако следващият не направи успешно резервно копие на данните и лидерът отново затвори, съобщението ще бъде загубено.
(3) acks=всички: Това означава, че лидерът трябва да изчака всички резервни копия, за да напишат логовете успешно, и тази стратегия ще гарантира, че данните няма да бъдат загубени, докато едно резервно копие оцелее. Това е най-силната гаранция.
(4) Възможни са и други настройки, като acks=2, които ще изискват определен брой acks, но тази стратегия обикновено рядко се използва. | | buffer.memory | Дълго | 33554432 | високо | Производителят може да се използва за кеширане на размера на паметта на данните. Ако данните се генерират по-бързо, отколкото се изпращат към брокера, производителят ще блокира или хвърли изключение, посочено с "block.on.buffer.full".
Тази настройка ще бъде свързана с общата памет, която производителят може да използва, но не е твърдо ограничение, тъй като не цялата памет, използвана от производителя, се използва за кеширане. Допълнителна памет се използва за компресия (ако се въведе компресия), а част – за заявки за поддръжка. | | compression.type | Низ | Никой | високо | Producer е видът компресия, използвана за компресиране на данни. По подразбиране е некомпресиран. Правилните стойности на опциите са none, gzip, snappy.
Компресията е най-добра за пакетна обработка, колкото повече съобщения се обработват на партиди, толкова по-добра е производителността на компресията. | | Повторни опити | int | 0 | високо | Задаването на стойност по-голяма от 0 ще накара клиента да изпрати отново всички данни, след като тези данни се провалят. Обърнете внимание, че тези повторни опити не се различават от тези, когато клиентът получи грешка при изпращане. Позволява повторни опити за евентуално промяна на реда на данните, ако и двата записа са изпратени към един и същ дял, първото съобщение се провали, а второто съобщение се появи по-рано от първото. | | batch.size | int | 16384 | Средно | Продуцентът ще се опита да групира записите със съобщения, за да намали броя на заявките. Това ще подобри производителността между клиент и сървър. Тази конфигурация контролира по подразбиране броя байтове на пакетните съобщения.
Не се правят опити за обработка на байтове на съобщенията над този брой байтове.
Заявките, изпратени към брокери, ще съдържат няколко партиди, които ще съдържат по една заявка за всеки дял.
По-малките стойности на партидиране се използват по-малко и могат да намалят пропускателната способност (0 използва само пакетна обработка). По-големите партидни стойности губят повече място в паметта, затова трябва да отделяте памет за конкретни стойности на партида. | | client.id | Низ | | Средно | Този низ се изпраща към сървъра при направене на заявка. Целта е да се проследи източникът на заявките, за да се позволи на някои приложения извън списъка с разрешени IP/портове да изпращат информация. Това приложение може да задава всяка струна, защото няма друга функционална функция освен запис и проследяване | | linger.ms | Дълго | 0 | Средно | Продуцентската група ще агрегира всички съобщения, които пристигнат между заявката и изпращането, като записва отделна партида заявки. Обикновено това се случва само когато записът се генерира по-бързо от скоростта на изпращане. Въпреки това, при определени условия клиентът ще иска да намали броя на заявките или дори до умерено натоварване. Тази настройка ще се осъществи чрез добавяне на малко забавяне – т.е. вместо да изпрати запис веднага, продуцентът ще изчака определено време за забавяне, за да позволи изпращане на други записи на съобщения, които могат да бъдат пакетирани. Това може да се счита за подобен алгоритъм на TCP Nagle. Тази настройка задава по-висока граница на латентност при групиране: след като получим batch.size на дяла, той ще го изпрати веднага независимо от тази настройка, но ако получим съобщение с много по-малък брой байтове от тази, трябва да "задържим" определено време, за да получим повече съобщения. Тази настройка по подразбиране е 0, т.е. няма забавяне. Настройката linger.ms=5, например, ще намали броя на заявките, но същевременно увеличава забавянето с 5ms. | | max.request.size | int | 1028576 | Средно | Максималният брой поискани байтове. Това е и ефективно покритие за максималния записан размер. Забележка: Сървърът има собствено управление на размерите на записите на съобщения, което се различава от тази настройка. Тази настройка ограничава броя заявки, които производителите могат да изпращат на едро наведнъж, за да се предотврати голям брой заявки. | | receive.buffer.bytes | int | 32768 | Средно | Размерът на приемащия кеш на TCP, който се използва при четене на данни | | send.buffer.bytes | int | 131072 | Средно | Размерът на кеша за изпращане на TCP, който се използва при изпращане на данни | | timeout.ms | int | 30000 | Средно | Тази конфигурационна опция контролира максималното време, през което сървърът чака потвърждение от последователи. Ако броят на потвърдените заявки не бъде изпълнен в този период, се връща грешка. Този лимит на тайм-аут се измерва от страна на сървъра и няма мрежова латентност, включително заявки | | block.on.buffer.full | булева | true | нисък | Когато кешът ни в паметта свърши, трябва да спрем да получаваме нови записи на съобщения или да хвърляме грешки. По подразбиране това е настроено на true, но някои блокирания може да не си струва да се очакват, затова е по-добре да хвърлите грешка веднага. Това е така, когато е настроено на false: продуцентът хвърля грешка за изключение: BufferExhaustedException, ако записът е изпратен и кешът е пълен | | metadata.fetch.timeout.ms | Дълго | 60000 | нисък | Той се отнася до данни за първи път на някои елементи, които сме получили. Елементите включват: тема, хост, дялове. Тази конфигурация се отнася до времето, необходимо елементът да завърши успешно според fetch, в противен случай ще бъде изпратено изключение на клиента. | | metadata.max.age.ms | Дълго | 300000 | нисък | Времето в микросекунди е интервалът, на който принуждаваме метаданните да бъдат актуализирани. Дори и да не видим промени в ръководството на разделението. | | metric.reporters | Списък | [] | нисък | Списък с класове, използвани за измерване на метрики. Внедряването на интерфейса MetricReporter ще позволи добавяне на класове, които се променят при генериране на нови метрики. JmxReporter винаги ще включва начин за регистриране на JMX статистики | | metrics.num.samples | int | 2 | нисък | Броят на пробите, използвани за поддържане на метрики | | metrics.sample.window.ms | Дълго | 30000 | нисък | Системата за метрики поддържа конфигурируем брой проби в коригиращ размер на прозореца. Тази конфигурация конфигурира размера на прозореца, например. Може да поддържаме две проби за продължителност 30 секунди. Когато прозорецът се разтегне, изтриваме и пренаписваме най-стария прозорец | | recoonect.backoff.ms | Дълго | 10 | нисък | Когато връзката се провали, времето за изчакване, когато се свържем отново. Това предотвратява многократни свързвания на клиента | | retry.backoff.ms | Дълго | 100 | нисък | Времето за изчакване преди да опитам да опитам отново неуспешна заявка за продукция. Избягвайте да се засядате в цикъл на изпращане-провал и мъртъв.
|
|
Публикувано в 24.06.2021 г. 11:47:12 ч.
|
|
|
Хазяин