mindol1004/excel-download
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

#15 kafka : producer - callback producer

Browse Source
This commit is contained in:
haerong22
2022-08-02 01:43:31 +09:00
parent b9a344b361
commit 664f5bfc7a
4 changed files with 68 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
kafka/kafka-producer/src/main/java/KafkaConfig.java
View File

@@ -13,6 +13,7 @@ public class KafkaConfig {
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfig.BOOTSTRAP_SERVERS);
configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// configs.put(ProducerConfig.ACKS_CONFIG, 0);
return new KafkaProducer<>(configs);
}

16
kafka/kafka-producer/src/main/java/ProducerCallback.java Normal file
View File

@@ -0,0 +1,16 @@
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ProducerCallback implements Callback {
private final static Logger logger = LoggerFactory.getLogger(ProducerCallback.class);
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null)
logger.error(e.getMessage(), e);
else
logger.info(recordMetadata.toString());
}
}

21
kafka/kafka-producer/src/main/java/ProducerWithAsyncCallback.java Normal file
View File

@@ -0,0 +1,21 @@
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class ProducerWithAsyncCallback {
public static void main(String[] args) {
KafkaProducer<String, String> producer = KafkaConfig.setup();
ProducerRecord<String, String> record =
new ProducerRecord<>(KafkaConfig.TOPIC_NAME, "async-callback", "async-callback");
producer.send(record, new ProducerCallback());
producer.flush();
producer.close();
}
}

30
kafka/kafka-producer/src/main/java/ProducerWithSyncCallback.java Normal file
View File

@@ -0,0 +1,30 @@
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
public class ProducerWithSyncCallback {
private final static Logger logger = LoggerFactory.getLogger(ProducerWithSyncCallback.class);
public static void main(String[] args) {
KafkaProducer<String, String> producer = KafkaConfig.setup();
ProducerRecord<String, String> record =
new ProducerRecord<>(KafkaConfig.TOPIC_NAME, "sync-callback", "sync-callback");
try {
RecordMetadata metadata = producer.send(record).get();
logger.info(metadata.toString());
} catch (Exception e) {
logger.error(e.getMessage(),e);
} finally {
producer.flush();
producer.close();
}
}
}