public KafkaClient(String hosts, String group, String topic) {
HashMap<String, Object> configs = new HashMap<>();
configs.put("bootstrap.servers", hosts);
configs.put("group.id", group);
configs.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
configs.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
this.producer = new KafkaProducer<>(configs);
log.debug("create kafka client. ");
this.gson = new Gson();
}
public void sendMessage(String data) {
String jsonMsg = gson.toJson(new Message(data, new Date().getTime()));
log.debug("sendMessage [{}]", jsonMsg);
producer.send(new ProducerRecord<>(topic, jsonMsg));
}
'java' 카테고리의 다른 글
JUnit test Mock 객체 지연 셋팅 (0) | 2024.06.28 |
---|---|
java.util.concurrent.CompletionException: java.lang.IllegalStateException: Duplicate key (0) | 2022.03.25 |
spring-batch 에서 kafka 사용할때 ConcurrentModificationException 발생시 (0) | 2021.11.25 |
springboot- hadoop hive 설정 (0) | 2021.11.12 |
gson 필드에 어노테이션으로 필드 지정 (0) | 2021.09.13 |