Kafka生产者Java客户端

Li_XiaoJin

发布于 2022-06-10 13:46:41

39000

代码可运行

文章被收录于专栏：Lixj's BlogLixj's Blog

运行总次数：0

代码可运行

记录，避免重复造轮子。

@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";