Конфігурації брокера
| Власність | За промовчанням | Опис | | broker.id | | Кожен брокер можна ідентифікувати за унікальним невід'ємним цілочисельним ідентифікатором; Цей id може використовуватися як «ім'я» брокера, а його існування дозволяє брокеру мігрувати на інший хост/порт, не плутаючи споживачів. Ви можете обрати будь-яке число як ідентифікатор, якщо ID унікальний. | | log.dirs | /tmp/kafka-logs | Шлях, де Кафка зберігає дані. Цей шлях не є унікальним, він може бути кільком, і шляхи потрібно розділяти лише комами; Щоразу, коли створюється нова частина, вона обирається для проходження шляху, що містить найменшу кількість розділень. | | Порт | 6667 | Сервер приймає порт, до якого підключається клієнт | | zookeeper.connect | нуль | Формат рядка з'єднання ZooKeeper: hostname:port, де hostname та port — це хост і порт вузла в кластері ZooKeeper відповідно. Щоб підключатися до інших вузлів ZooKeeper, коли хост виходить з ладу, ви можете створити кілька хостів наступним чином:
hostname1:port1, hostname2:port2, hostname3:port3.
ZooKeeper дозволяє додати шлях «chroot» для зберігання всіх даних Kafka у кластері під певним шляхом. Коли кілька кластерів Kafka або інших додатків використовують один і той самий кластер ZooKeeper, цей метод можна використати для встановлення шляху зберігання даних. Це можна реалізувати, відформатувавши рядок з'єднання таким чином:
hostname1:port1,hostname2:port2,hostname3:port3/chroot/path
Ця конфігурація зберігає всі дані кластера Kafka у шляху /chroot/path. Зверніть увагу, що перед запуском брокера потрібно створити цей шлях, а споживачі повинні використовувати той самий формат підключення. | | message.max.bytes | 1000000 | Максимальний розмір повідомлень, який сервер може отримати. Важливо, щоб налаштування споживача і виробника щодо цієї властивості були синхронізовані, інакше повідомлення, опубліковане виробником, буде надто великим для споживача. | | num.network.threads | 3 | Кількість мережевих потоків, які використовує сервер для обробки мережевих запитів; Зазвичай змінювати цю нерухомість не потрібно. | | num.io.threads | 8 | Кількість потоків введення/виведення, які використовує сервер для обробки запитів; Кількість потоків має дорівнювати принаймні кількості жорстких дисків. | | background.threads | 4 | Кількість потоків, що використовуються для фонової обробки, наприклад, видалення файлів; Вам не потрібно змінювати цю власність. | | queued.max.requests | 500 | Максимальна кількість запитів, які можна поставити в чергу для обробки потоків введення/виведення до того, як мережевий потік припинить читати нові запити. | | host.name | нуль | ім'я брокера; Якщо ім'я хоста вже встановлене, брокер буде прив'язаний лише до цієї адреси; Якщо налаштування немає, він прив'язується до всіх інтерфейсів і публікує одну копію на ZK | | advertised.host.name | нуль | Якщо встановлено, він надсилається виробникам, споживачам та іншим брокерам як хост-ім'я брокера | | реклама.порт | нуль | Цей порт надається виробникам, споживачам та іншим брокерам і використовується для встановлення зв'язку; Його потрібно встановити лише якщо фактичний порт і порт, який сервер має прив'язати, різні. | | 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.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.s | Ніхто | Максимальна кількість введень/операцій, яку може мати прибиральник журналів під час ущільнення журналу. Це налаштування обмежує прибиральника, щоб не заважати активним сервісам запитів. | | 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.байти | 10*1024*1024 | Максимальний розмір на сегмент колоди. Зверніть увагу, що якщо розмір log досягає цього значення, потрібно згенерувати новий сегмент log, навіть якщо розмір не перевищує ліміт log.segment.bytes. | | log.index.interval.bytes | 4096 | Під час вилучення потрібно просканувати найближчий зсув із певним простором, чим більший настрій, тим краще, зазвичай використовуйте стандартне значення | | 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 | Controller-to-broker-channles | | default.replication.factor | 1 | За замовчуванням кількість резервних копій стосується лише автоматично створених тем | | replica.lag.time.max.ms | 10000 | Якщо послідовник не надішле запит на доставку протягом цього часу, лідер видаляє його з ISR і визнає його підвішеним | | replica.lag.max.повідомлення | 4000 | Якщо репліка має більше незахищених копій, лідер видаляє послідовника і вважає його підвішеним | | replica.socket.timeout.ms | 30*1000 | Час тайм-ауту лідера для запитів мережі сокетів при резервному копіюванні даних | | replica.socket.receive.buffer.bytes | 64*1024 | Буфер прийому сокета при надсилання мережевого запиту лідеру під час резервного копіювання | | replica.fetch.max.байтів | 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 | Запит на отримання інтервалу очищення | | 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,години} | 0 | Максимальна кількість тремтіння, абстрагованої з logRollTimeMillis | | num.recovery.threads.per.data.dir | 1 | Кількість потоків, які кожен каталог даних використовує для ведення відновлення | | нечистий.лідер.вибор.enable | 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.retention.ms | 86400000 (24 години) | log.cleaner.delete.retention.ms | Різниця між log.retention.minutes полягає в тому, що один керує нестисненими даними, а інший — стисненими. Цю конфігурацію можна замінити параметрами закріплення при створенні теми | | Змити.повідомлення | Ніхто | log.flush.interval.messages | Ця конфігурація визначає часовий інтервал для примусового запуску fsync-журналів. Наприклад, якщо ця опція встановлена на 1, то після кожного повідомлення потрібен fsync, а якщо встановлено на 5 — fsync для кожних 5 повідомлень. Загалом, рекомендується не встановлювати це значення. Встановлення цього параметра вимагає необхідного компромісу між «надійністю даних» і «продуктивністю». Якщо це значення занадто велике, це спричинятиме тривалий час «fsync» кожного разу (блокування введення), а якщо це значення занадто мале — призведе до великої кількості «fsync», що також означає певну затримку у загальному запиті клієнта. Фізична несправність сервера призведе до втрати повідомлень без fsync. | | flush.ms | Ніхто | log.flush.interval.ms | Ця конфігурація використовується для фіксації інтервалу часу між примусом fsync-журналів на диск; Наприклад, якщо встановити на 1000, то fsync потрібен кожні 1000 мс. Цей варіант зазвичай не рекомендується | | index.interval.bytes | 4096 | log.index.interval.bytes | Стандартне налаштування гарантує, що ми додаємо індекс до повідомлення кожні 4096 байтів, а більше індексів робить повідомлення читання ближчим, але розмір індексу збільшується. Ця опція зазвичай не є обов'язковою | | max.message.bytes | 1000000 | max.message.bytes | Максимальний розмір повідомлення додавання Кафки. Зверніть увагу, що якщо ви збільшуєте цей розмір, вам також потрібно збільшити розмір завантаження користувача, щоб користувач міг отримувати повідомлення до цих максимальних розмірів. | | min.cleanable.dirty.ratio | 0.5 | min.cleanable.dirty.ratio | Ця конфігурація контролює, як часто компресор для дерев намагається очистити колоди. За замовчуванням уникаються логи зі ступенем стиснення понад 50%. Це співвідношення дозволяє уникнути найбільшої втрати простору | | min.insync.replicas | 1 | min.insync.replicas | Коли виробник встановлений на request.required.acks у -1, min.insync.replicas вказує мінімальну кількість реплік (кожен запис repica має бути успішним), і якщо ця кількість не досягнута, виробник створить виняток. | | retention.bytes | Ніхто | log.retention.bytes | Якщо ви використовуєте політику утримання "видалити", ця конфігурація відображає максимальний розмір, якого може досягти журнал до його видалення. За замовчуванням немає обмежень по розміру, а лише часовий ліміт | | retention.ms | 7 днів | log.retention.minutes | Якщо ви використовуєте політику утримання "видалення", ця конфігурація стосується часу, коли журнал було збережено до журналу видалення. | | segment.bytes | 1GB | log.segment.bytes | У Kafka логарифмічні журнали зберігаються чанками, і ця конфігурація стосується розміру логарифмічних журналів, поділених на шматки | | segment.index.bytes | 10 МБ | log.index.size.max.байти | Ця конфігурація приблизно відповідає розміру індексного файлу, відображеного між зміщеннями та розташуванням файлів; Цю конфігурацію зазвичай не потрібно змінювати | | segment.ms | 7 днів | log.roll.hours | Навіть якщо файл log chunk не досягає потрібного розміру, який потрібно видалити або стиснути, коли час журналу досягне цієї верхньої межі, буде примусово створювати новий файл log chunk | | segment.jitter.ms | 0 | log.roll.jitter. {MS,години} | Максимальний джитер для відняття від logRollTimeMillis. |
Споживчі конфігурації
| Власність | За промовчанням | Опис | | group.id | | Рядок, який унікально ідентифікує групу, в якій розташований споживчий процес, і якщо встановлено той самий ідентифікатор групи, це означає, що ці процеси належать одній групі споживача | | zookeeper.connect | | Вкажіть рядок з'єднання Zookeeper, формат — hostname:port, тут хост і порт є одночасно хостом і портом сервера Zookeeper, щоб уникнути втрати зв'язку після виходу з ладу комп'ютера Zookeeper, можна вказати кілька hostname:ports, використовуючи коми як розділення:
Ім'я хоста1:порт1,ім'я хоста2:порт2,ім'я хоста3:порт3
Ви можете додати шлях chroot ZooKeeper до рядка з'єднання ZooKeeper, яка використовується для зберігання власних даних, таким чином:
hostname1:port1,hostname2:port2,hostname3:port3/chroot/path | | consumer.id | нуль | Налаштування не потрібне, і зазвичай він генерується автоматично | | 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 | Кількість потоків fetcher для завантаження даних | | auto.commit.enable | true | Якщо це правда, зсув повідомлення, отриманого споживачем, автоматично синхронізується з доглядачем зоопарку. Цей коміт-офсет буде використаний новим споживачем, коли процес завершиться | | auto.commit.interval.ms | 60*1000 | Частота, з якою споживач надсилає офсет доглядачу зоопарку, становить секунди | | queued.max.message.chunks | 2 | Максимальна кількість повідомлень, що використовуються для кешування для споживання. Кожен блок має бути таким самим, як 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 немає ініціалізованого зсуву, якщо зсув є відповіддю на наступне значення:
найменший: автоматичне скидання зсуву на найменший зсув
найбільший: Автоматичне скидання зсуву до зсуву найбільшого
Все інше: Це виняток для споживача | | 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). Це обов'язково при переході з офсетного сховища на основі Zookeeper на Kafka-базове офсетне зберігання. Для будь-якої групи споживачів безпечно вимкнути цю опцію після завершення міграції | | partition.assignment.strategy | Ареал | Обирайте між політиками «діапазон» і «roundrobin» як політикою призначення розділів для споживчих потоків даних; Круглий розподільник розділів розподіляє всі доступні розділи, а також усі доступні споживчі потоки. Він призначає цикл розділів споживчому потоку. Якщо всі споживчі екземпляри підписані на визначений, розбиття поділяються на детерміновані дистрибутиви. Стратегія розподілу за круговим етапом можлива лише за умови виконання наступних умов: (1) Кожна тема має однакову кількість потоків даних на рівень споживача. (2) Колекція підписаних тем визначається для кожного споживчого екземпляра в споживчій групі. |
Конфігурації продюсера
| Власність | За промовчанням | Опис | | metadata.broker.list | | Подавайте підготовку. 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) синхронізація: Синхронізоване надсилання
Встановивши виробник на асинхронний, ми можемо обробляти запити пакетами (що добре для більшої пропускної здатності), але це також створює ймовірність втрати невідправлених даних клієнтом | | serializer.class | kafka.serializer.DefaultEncoder | Категорія серіалізації повідомлення. Стандартний енкодер вводить один байт[] і повертає той самий байт[] | | key.serializer.class | | Клас серіалізації для ключових слів. Якщо це не вказано, за замовчуванням збігається з повідомленням | | partitioner.class | kafka.producer.DefaultPartitioner | клас partitioner для поділу повідомлень між підтемами. Стандартний розділ базується на хеш-таблиці ключа | | compression.codec | Ніхто | Цей параметр може встановити кодек для стиснення даних, який можна вибрати як «none», «gzip», «snappy». | | compressed.topics | нуль | Цей параметр можна використати для визначення, чи стискаються певні теми. Якщо стиснутий кодек є кодеком, відмінним від 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, повідомлення в межах 100 мс обробляються пакетами. Це покращить пропускну здатність, але збільшить затримку через кешування. | | queue.buffering.max.messages | 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=all: Це означає, що лідеру потрібно чекати, поки всі резервні копії успішно запишуть журнали, і ця стратегія гарантує, що дані не будуть втрачені, доки одна з них зберігається. Це найсильніша гарантія.
(4) Можливі й інші налаштування, такі як acks=2, які вимагатимуть певної кількості acks, але ця стратегія зазвичай використовується рідко. | | buffer.memory | довгий | 33554432 | високий | Виробник може використовуватися для кешування розміру пам'яті даних. Якщо дані генеруються швидше, ніж надсилаються брокеру, виробник блокує або викидає виняток, позначений як "block.on.buffer.full".
Це налаштування буде пов'язане з загальною пам'яттю, яку може використовувати виробник, але це не є жорстким обмеженням, оскільки не вся пам'ять, яку використовує виробник, використовується для кешування. Додаткова пам'ять використовується для стиснення (якщо впроваджено стиснення), а частина — для запитів на обслуговування. | | compression.type | Рядок | Ніхто | високий | Producer — це тип стиснення, який використовується для стиснення даних. За замовчуванням він не стиснений. Правильні значення опції: none, gzip, snappy.
Компресію найкраще використовувати для пакетної обробки, чим більше повідомлень обробляється пакетами, тим краща продуктивність стиснення. | | Повторні спроби | int | 0 | високий | Встановлення значення більше за 0 змусить клієнта повторно надіслати будь-які дані після відмови. Зверніть увагу, що ці повторні спроби не відрізняються від тих, що випадають на помилку відправлення клієнтом. Дозволяє повторні спроби потенційно змінити порядок даних, якщо обидва записи повідомлень надсилаються на один розділ, перше повідомлення не спрацює, друге з'являється раніше за перше. | | Розмір партії | int | 16384 | Середнє | Продюсер намагатиметься групувати записи повідомлень, щоб зменшити кількість запитів. Це покращить продуктивність між клієнтом і сервером. Ця конфігурація контролює кількість байтів пакетної обробки за замовчуванням.
Жодних спроб обробляти байти повідомлення, що перевищують цю кількість байтів.
Запити, надіслані брокерам, містять кілька пакетів, які міститимуть по одному запиту для кожного розділу.
Менші пакетні значення використовуються рідше і можуть зменшити пропускну здатність (0 використовуватиме лише пакетну обробку). Більші пакетні значення витрачають більше пам'яті, тому потрібно виділяти пам'ять для конкретних пакетних значень. | | client.id | Рядок | | Середнє | Цей рядок надсилається на сервер при поданні запиту. Мета полягає в тому, щоб мати змогу відстежити джерело запитів, щоб дозволити деяким додаткам поза списком дозволів IP/порту надсилати інформацію. Цей додаток може встановлювати будь-який рядок, оскільки він не має функціональної функції, окрім запису та відстеження | | linger.ms | довгий | 0 | Середнє | Група продюсерів агрегує будь-які повідомлення, що надходять між запитом і відправленням, записуючи окрему партію запитів. Зазвичай це відбувається лише тоді, коли запис генерується швидше за швидкість відправлення. Однак за певних умов клієнт захоче зменшити кількість запитів або навіть до помірного навантаження. Це налаштування здійснюється шляхом додавання невеликої затримки — тобто замість одразового надсилання платівки продюсер чекатиме певного часу затримки, щоб дозволити надіслати інші записи повідомлень, які можна буде пакетувати. Це можна вважати подібним алгоритмом до TCP Nagle. Це налаштування встановлює вищу межу затримки для пакетування: як тільки ми отримуємо розмір batch.size розділу, він надсилає його одразу незалежно від цього параметра, але якщо ми отримуємо повідомлення з набагато меншою кількістю байтів, ніж це налаштування, нам потрібно «затриматися» на певний час, щоб отримати більше повідомлень. Це налаштування за замовчуванням становить 0, тобто затримки немає. Наприклад, встановлення linger.ms=5 зменшує кількість запитів, але водночас збільшує затримку на 5 мс. | | 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 | низький | Вона стосується даних першого разу деяких елементів, які ми отримали. Елементи включають: тему, хост, розділи. Ця конфігурація стосується часу, необхідного для успішного завершення елемента відповідно до завантаження, інакше клієнту буде надіслано виняток. | | 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
|
|
|
Орендодавець