Kafka生产者Java客户端
Kafka生产者Java客户端
Li_XiaoJin
发布于 2022-06-10 21:46:41
发布于 2022-06-10 21:46:41
?
记录，避免重复造轮子。
@Service
@Slf4j
public class KafkaCommonProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
/**
* 发送消息
* @param topic topic
* @param t 消息
* @param <T> 消息类型
* @return 发送结果
*/
public <T> ListenableFuture<SendResult<String, String>> send(String topic, T t) {
final String message = JSON.toJSONString(t);
return kafkaTemplate.send(topic, message);
}
/**
* 发送消息
* @param topic topic
* @param key key
* @param t 消息
* @param <T> 消息类型
* @return 发送结果
*/
public <T> ListenableFuture<SendResult<String, String>> send(String topic, String key, T t) {
final String message = JSON.toJSONString(t);
return kafkaTemplate.send(topic, key, message);
}
}如果需要回调则可以
public void send(String warningMessage) {
log.info(">>>>> Kafka消息发送，topic: {}, Key: {}, message: {}", TOPIC_NAME, TOPIC_NAME, warningMessage);
ListenableFuture<SendResult<String, String>> future = kafkaCommonProducer.send(TOPIC_NAME, TOPIC_NAME, warningMessage);
future.addCallback(
success -> log.info(">>>>> Kafka消息发送成功,{}", success.toString()),
failure -> log.info(">>>>> Kafka消息发送失败,{}", failure.getMessage())
);
}application.yml 配置如下：
spring:
application:
name: test-kafka-msg
kafka:
bootstrap-servers: localhost:20902
############如果采用SASL认证的话需要添加以下内容
producer:
properties:
sasl.mechanism: SCRAM-SHA-256
security.protocol: SASL_PLAINTEXT
sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="test" password="test";Copyright: 采用 知识共享署名4.0 国际许可协议进行许可 Links: /developer/article/2020541
?
本文参与?腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2021-07-10，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读