Kafka-生产商批次数

Question

有没有办法确定Kafka制作人为一组特定消息创建的批次数？例如，如果我在一个循环中发送10K条消息，有没有办法检查发送了多少批？我将“batch.size”设置为一个高值，我的期望是消息将被缓冲，并且在我的消费者中看到消息时会有延迟。然而，这似乎是打印几乎立即在我的消费者计划。

批处理时的默认值。尺寸是16384。这是字节数吗？

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;


public class KafkaProducerApp {

    public static void main(String[] args){
        Properties properties = new Properties();
        properties.put("bootstrap.servers","localhost:9092,localhost:9093,localhost:9094");
        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("acks","0");
        properties.put("batch.size",33554432);

        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);
        Map<Integer,Integer> partitionCount = new HashMap<Integer,Integer>();
        partitionCount.put(0,0);
        partitionCount.put(1,0);
        partitionCount.put(2,0);


        try{
            Date from = new Date();
            for(int i=0;i<10000;i++) {
                RecordMetadata ack = kafkaProducer.send(new ProducerRecord<String, String>("test_topic", Integer.toString(i), "MyMessage" + Integer.toString(i))).get();
                //RecordMetadata ack = kafkaProducer.send(new ProducerRecord<String,String>("test_topic",0,Integer.toString(i), "MyMessage" + Integer.toString(i))).get();
                System.out.println(" Offset = " + ack.offset());
                System.out.println(" Partition = " + ack.partition());
                partitionCount.put(ack.partition(),partitionCount.get(ack.partition())+1);

            }
            Date to = new Date();
            System.out.println(" partition 0 =" + partitionCount.get(0));
            System.out.println(" partition 1 =" + partitionCount.get(1));
            System.out.println(" partition 2 =" + partitionCount.get(2));
            System.out.println(" Elapsed Time = " + (to.getTime()-from.getTime())/1000);

        } catch (Exception ex){
            ex.printStackTrace();
        } finally {
            kafkaProducer.close();
        }



    }

}

Answer 1

您需要的是产品请求的总数。

您可以使用JMX Mbeankafka.producer查看每秒产生请求的平均数量：type=生产者-指标，client-id=（[-. w]）