Kafka 소스 해독-생산자KafkaProducer

Kafka 소스 해독-생산자KafkaProducer(중단 지속 더...)

//  producer 
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
//  
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer producer = new KafkaProducer<>(props);
for(int i = 0; i < 100; i++)
    producer.send(new ProducerRecord("my-topic", Integer.toString(i), Integer.toString(i)));

producer.close();

매개 변수

    private static final Logger log = LoggerFactory.getLogger(KafkaProducer.class);
    private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
    private static final String JMX_PREFIX = "kafka.producer";
    // client.id
    private String clientId;
    private final Partitioner partitioner;
    private final int maxRequestSize;
    private final long totalMemorySize;
    //  Kafka 
    private final Metadata metadata;
    private final RecordAccumulator accumulator;
    private final Sender sender;
    private final Metrics metrics;
    // i/o ， 
    private final Thread ioThread;
    private final CompressionType compressionType;
    private final Sensor errors;
    //  
    private final Time time;
    private final Serializer keySerializer;
    private final Serializer valueSerializer;
    //  
    private final ProducerConfig producerConfig;
    private final long maxBlockTimeMs;
    private final int requestTimeoutMs;
    private final ProducerInterceptors interceptors;

메서드

 byte[] key = "key".getBytes();
 byte[] value = "value".getBytes();
 ProducerRecord record = new ProducerRecord("my-topic", key, value)
 producer.send(record).get();

ProducerRecord record = new ProducerRecord("the-topic", key, value);
 producer.send(myRecord,
               new Callback() {
                   public void onCompletion(RecordMetadata metadata, Exception e) {
                       if(e != null) {
                          e.printStackTrace();
                       } else {
                          System.out.println("The offset of the record we just sent is: " + metadata.offset());
                       }
                   }
               });

producer.send(new ProducerRecord(topic, partition, key1, value1), callback1);
producer.send(new ProducerRecord(topic, partition, key2, value2), callback2);