Java, Cách nhận số lượng tin nhắn trong một chủ đề trong apache kafka

Question

Tôi đang sử dụng apache kafka để nhắn tin. Tôi đã triển khai nhà sản xuất và người tiêu dùng trong Java. Làm thế nào chúng ta có thể nhận được số lượng tin nhắn trong một chủ đề?

Answer 1

Cách duy nhất để suy nghĩ về điều này từ quan điểm của người tiêu dùng là thực sự tiêu thụ các tin nhắn và đếm chúng sau đó.

Nhà môi giới Kafka cho thấy các bộ đếm JMX đối với số lượng tin nhắn nhận được kể từ khi khởi động nhưng bạn không thể biết có bao nhiêu thông báo đã bị xóa.

Trong hầu hết các trường hợp phổ biến, thư trong Kafka được xem là luồng vô hạn và nhận được giá trị rời rạc về số lượng hiện đang lưu trên đĩa không liên quan. Hơn nữa, mọi thứ trở nên phức tạp hơn khi giao dịch với một nhóm các nhà môi giới, tất cả đều có một tập hợp con các thông điệp trong một chủ đề.

Answer 2

Tôi chưa thử tự mình this nhưng có vẻ hợp lý.

Bạn cũng có thể sử dụng kafka.tools.ConsumerOffsetChecker (source).

Answer 3

Nó không phải là java, nhưng có thể hữu ích

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell 
    --broker-list <broker>: <port> 
    --topic <topic-name> --time -1 --offsets 1 
    | awk -F ":" '{sum += $3} END {print sum}' 

Answer 4

Sử dụng https://prestodb.io/docs/current/connector/kafka-tutorial.html

Một động cơ SQL siêu, được cung cấp bởi Facebook, kết nối trên nhiều nguồn dữ liệu (Cassandra, Kafka, JMX, Redis. ..). PrestoDB đang chạy như một máy chủ với công nhân tùy chọn (có một chế độ độc lập mà không cần thêm công nhân), sau đó bạn sử dụng một JAR thực thi nhỏ (được gọi là presto CLI) để thực hiện truy vấn.

Một khi bạn đã cấu hình tốt máy chủ Presto, bạn có thể sử dụng SQL traditionnal:

SELECT count(*) FROM TOPIC_NAME; 

Answer 5

Tôi thực sự sử dụng này cho benchmarking POC tôi. Mục bạn muốn sử dụng ConsumerOffsetChecker. Bạn có thể chạy nó bằng cách sử dụng kịch bản bash như dưới đây.

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --topic test --zookeeper localhost:2181 --group testgroup 

Và dưới đây là kết quả: Như bạn có thể nhìn thấy trên hộp màu đỏ, 999 là số của thông điệp hiện trong chủ đề này.

Cập nhật: ConsumerOffsetChecker không được chấp nhận kể từ 0.10.0, bạn có thể muốn bắt đầu sử dụng ConsumerGroupCommand.

Answer 6

Apache Kafka lệnh để có được un xử lý thông điệp trên tất cả các phân vùng của một chủ đề:

kafka-run-class kafka.tools.ConsumerOffsetChecker 
    --topic test --zookeeper localhost:2181 
    --group test_group 

Prints:

Group  Topic  Pid Offset   logSize   Lag    Owner 
test_group test   0 11051   11053   2    none 
test_group test   1 10810   10812   2    none 
test_group test   2 11027   11028   1    none 

Cột 6 là những thông điệp un-xử lý. Thêm chúng lên như thế này:

kafka-run-class kafka.tools.ConsumerOffsetChecker 
    --topic test --zookeeper localhost:2181 
    --group test_group 2>/dev/null | awk 'NR>1 {sum += $6} 
    END {print sum}' 

awk đọc hàng, bỏ qua dòng tiêu đề và thêm cột thứ 6 vào cuối bản in tổng.

Prints

5 

Answer 7

Để có được tất cả các tin nhắn được lưu trữ cho chủ đề, bạn có thể tìm kiếm khách hàng để đầu và kết thúc của dòng cho mỗi phân vùng và tổng hợp các kết quả

List<TopicPartition> partitions = consumer.partitionsFor(topic).stream() 
     .map(p -> new TopicPartition(topic, p.partition())) 
     .collect(Collectors.toList()); 
    consumer.assign(partitions); 
    consumer.seekToEnd(Collections.emptySet()); 
Map<TopicPartition, Long> endPartitions = partitions.stream() 
     .collect(Collectors.toMap(Function.identity(), consumer::position)); 
    consumer.seekToBeginning(Collections.emptySet()); 
System.out.println(partitions.stream().mapToLong(p -> endPartitions.get(p) - consumer.position(p)).sum()); 

Answer 8

Sử dụng Ứng dụng khách Java của Kafka 2.11-1.0.0, bạn có thể thực hiện như sau:

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
    consumer.subscribe(Collections.singletonList("test")); 
    while(true) { 
     ConsumerRecords<String, String> records = consumer.poll(100); 
     for (ConsumerRecord<String, String> record : records) { 
      System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); 

      // after each message, query the number of messages of the topic 
      Set<TopicPartition> partitions = consumer.assignment(); 
      Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions); 
      for(TopicPartition partition : offsets.keySet()) { 
       System.out.printf("partition %s is at %d\n", partition.topic(), offsets.get(partition)); 
      } 
     } 
    } 

Đầu ra là s omething như thế này:

offset = 10, key = null, value = un 
partition test is at 13 
offset = 11, key = null, value = deux 
partition test is at 13 
offset = 12, key = null, value = trois 
partition test is at 13 

Answer 9

Trong hầu hết các phiên bản gần đây của Kafka Manager, có một cột với tiêu đề tóm tắt Hiệu số gần đây.