需求

我們需要通過Spring Kafka庫，將消息推送給Kafka的topic中。這里假設Kafka的集群和用戶我們都有了。這里Kafka認證采取SASL_PLAINTEXT方式接入，SASL 采用 SCRAM-SHA-256 方式加解密。

pom.xml

<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
</dependency>

我這里不需要寫版本號，應為我使用的Spring Boot。Spring Boot會自動幫我挑選spring-kafka應該使用哪個版本合適。

application.yml

spring:
  kafka:
    producer:
	  # kafka集群地址
      bootstrap-servers: xx.xx.xx.xxx:9092,xx.xx.xx.xxx:9092,xx.xx.xx.xxx:9092
	  client-id: producer-dev
      # SASL_PLAINTEXT 接入方式
      security:
        protocol: SASL_PLAINTEXT
      # 序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        # SASL 采用 SCRAM-SHA-256 方式
        sasl:
          mechanism: SCRAM-SHA-256
    # jaas配置
    jaas:
      options:
        username: kafkauser
        password: kafkapwd
      enabled: true
      login-module: org.apache.kafka.common.security.scram.ScramLoginModule
      control-flag: required

以上，是關于Spring Kafka的全部配置。下面摘要出來的配置，是可以單獨配置在配置中心的：

topic:
  # 接收消息的主題配置
  save: hello_kafka_topic
spring:
  kafka:
    producer:
      client-id: producer-dev
      # kafka集群地址
      bootstrap-servers: xx.xx.xx.xxx:9092,xx.xx.xx.xxx:9092,xx.xx.xx.xxx:9092
    # jaas配置
    jaas:
      options:
        username: kafkauser
        password: kafkapwd

Java

KafkaProducerService.java

public interface KafkaProducerService {

    /**
     * 轉(zhuǎn)發(fā)消息到kafka
     */
    void sendToKafka(String msg);

}

KafkaProducerServiceImpl.java

import cn.com.xxx.service.KafkaProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaSendCallback;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;

/**
 * 轉(zhuǎn)發(fā)消息到kafka
 */
@RefreshScope
@Slf4j
@Service
public class KafkaProducerServiceImpl implements KafkaProducerService {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * kafka接收消息的主題
     */
    @Value("${topic.save}")
    private String topic;


    @Override
    public void sendToKafka(String msg) {
        log.info(String.format("$$$$ => Producing message: %s", msg));

        ProducerRecord<String, String> recordKafka = new ProducerRecord<>(topic, msg);

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(recordKafka);
        future.addCallback(new KafkaSendCallback<String, String>() {

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("成功發(fā)消息:{}給Kafka:{}", msg, result);
            }

            @Override
            public void onFailure(KafkaProducerException ex) {
                log.error("發(fā)消息:{}給Kafka:{}", msg, recordKafka, ex);
            }
        });
    }
}

到這里為止Spring Kafka生產(chǎn)者所有配置就都可以了。這里使用的異步監(jiān)聽kafka回調(diào)的方式發(fā)送消息。

總結(jié)

這里使用Spring Kafka庫異回調(diào)步給Kafka消息。這里使用的Spring Kafka庫是老版本，所以，這里的使用的回調(diào)類是ListenableFuture類。文章來源地址http://www.zghlxwxcb.cn/news/detail-714816.html

參考：

• Spring for Apache Kafka2.8.3
• Spring for Apache Kafka

到了這里，關于Spring Kafka生產(chǎn)者實現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！