Техническая колонка:GitHub.com/Делайте это с душой/Особые…
В то же время, вы также можете обратить внимание на мой публичный аккаунт WeChat AlwaysBeta, вас ждет более интересный контент.
В Kafka после 0.9.0.0 появилось несколько новых изменений.Одним из них является добавление роли GroupCoordinator на стороне сервера.Еще одним крупным изменением является изменение смещения информации топика из предыдущего хранилища на zookeeper на специальное. в теме (__consumer_offsets).
Тестовая версия этой статьи: kafka_2.11-2.2.0
Начать Кафку
Фоновый постоянный режим с параметрами-daemon,как:
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
Укажите порт JMX для запуска, укажите jmx, вы можете легко контролировать кластер Kafka
JMX_PORT=9991 /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
остановить Кафку
/usr/local/kafka/bin/kafka-server-stop.sh
Topic
Создать тему
параметр--topicУкажите название темы,--partitionsукажите количество разделов,--replication-factorУкажите количество резервных копий:
/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Учтите, что если в конфигурационном файле server.properties указана директория Kafka на zookeeper, параметры также должны быть указаны, иначе он сообщит, что нет доступных брокеров (то же самое верно и для следующих команд), например:
/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181/kafka --replication-factor 1 --partitions 1 --topic test
Список всех тем
/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181
Просмотр темы
/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Увеличьте количество разделов темы
/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic test --partitions 5
Просмотр максимального или минимального значения смещения раздела, указанного в теме
Когда время равно -1, это означает максимальное значение, а когда оно равно -2, это означает минимальное значение:
/usr/local/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1 --broker-list 127.0.0.1:9092 --partitions 0
удалить тему
/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --topic test --delete
новости производства
/usr/local/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
Использование сообщений
начинать с нуля
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
начать с конца
Чтобы получить данные с конца, необходимо указать раздел:
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 0
назначенный раздел
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 0
Возьмите указанное число
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 0 --max-messages 1
Группа потребителей
Указать группу
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test -group test_group --from-beginning
Список групп потребителей
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
Просмотр сведений о группе
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --describe
вывод:
Consumer group 'test_group' has no active members.
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test 0 5 5 0 - - -
# CURRENT-OFFSET: 当前消费者群组最近提交的 offset，也就是消费者分区里读取的当前位置
# LOG-END-OFFSET: 当前最高水位偏移量，也就是最近一个读取消息的偏移量，同时也是最近一个提交到集群的偏移量
# LAG：消费者的 CURRENT-OFFSET 与 broker 的 LOG-END-OFFSET 之间的差距
Удалить тему в группе
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --topic test --delete
удалить группу
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --delete
лидер баланса
/usr/local/kafka/bin/kafka-preferred-replica-election.sh --bootstrap-server localhost:9092
Встроенный инструмент измерения давления
/usr/local/kafka/bin/kafka-producer-perf-test.sh --topic test --num-records 100 --record-size 1 --throughput 100 --producer-props bootstrap.servers=localhost:9092
выше
Адрес столбца: GitHub.com/Делайте это с душой/Особые…