這篇文章將為大家詳細(xì)講解有關(guān)如何使用Java實(shí)現(xiàn)Kafka生產(chǎn)者和消費(fèi)者的示例，小編覺(jué)得挺實(shí)用的，因此分享給大家做個(gè)參考，希望大家閱讀完這篇文章后可以有所收獲。

成都創(chuàng)新互聯(lián)是專業(yè)的安陸網(wǎng)站建設(shè)公司，安陸接單;提供成都網(wǎng)站設(shè)計(jì)、網(wǎng)站建設(shè),網(wǎng)頁(yè)設(shè)計(jì),網(wǎng)站設(shè)計(jì),建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行安陸網(wǎng)站開(kāi)發(fā)網(wǎng)頁(yè)制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛(ài)的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊(duì),希望更多企業(yè)前來(lái)合作!

Kafka簡(jiǎn)介

Kafka是由Apache軟件基金會(huì)開(kāi)發(fā)的一個(gè)開(kāi)源流處理平臺(tái)，由Scala和Java編寫。Kafka的目標(biāo)是為處理實(shí)時(shí)數(shù)據(jù)提供一個(gè)統(tǒng)一、高吞吐、低延遲的平臺(tái)。

方式一：kafka-clients

引入依賴

在pom.xml文件中，引入kafka-clients依賴：

  org.apache.kafka
  kafka-clients
  2.3.1

生產(chǎn)者

創(chuàng)建一個(gè)KafkaProducer的生產(chǎn)者實(shí)例：

@Configuration
public class Config {

  public final static String bootstrapServers = "127.0.0.1:9092";

  @Bean(destroyMethod = "close")
  public KafkaProducer kafkaProducer() {
    Properties props = new Properties();
    //設(shè)置Kafka服務(wù)器地址
    props.put("bootstrap.servers", bootstrapServers);
    //設(shè)置數(shù)據(jù)key的序列化處理類
    props.put("key.serializer", StringSerializer.class.getName());
    //設(shè)置數(shù)據(jù)value的序列化處理類
    props.put("value.serializer", StringSerializer.class.getName());
    KafkaProducer producer = new KafkaProducer<>(props);
    return producer;
  }
}

在Controller中進(jìn)行使用：

@RestController
@Slf4j
public class Controller {

  @Autowired
  private KafkaProducer kafkaProducer;

  @RequestMapping("/kafkaClientsSend")
  public String send() {
    String uuid = UUID.randomUUID().toString();
    RecordMetadata recordMetadata = null;
    try {
     //將消息發(fā)送到Kafka服務(wù)器的名稱為“one-more-topic”的Topic中
      recordMetadata = kafkaProducer.send(new ProducerRecord<>("one-more-topic", uuid)).get();
      log.info("recordMetadata: {}", recordMetadata);
      log.info("uuid: {}", uuid);
    } catch (Exception e) {
      log.error("send fail, uuid: {}", uuid, e);
    }
    return uuid;
  }
}

消費(fèi)者

創(chuàng)建一個(gè)KafkaConsumer的消費(fèi)者實(shí)例：

@Configuration
public class Config {

  public final static String groupId = "kafka-clients-group";
  public final static String bootstrapServers = "127.0.0.1:9092";

  @Bean(destroyMethod = "close")
  public KafkaConsumer kafkaConsumer() {
    Properties props = new Properties();
    //設(shè)置Kafka服務(wù)器地址
    props.put("bootstrap.servers", bootstrapServers);
    //設(shè)置消費(fèi)組
    props.put("group.id", groupId);
    //設(shè)置數(shù)據(jù)key的反序列化處理類
    props.put("key.deserializer", StringDeserializer.class.getName());
    //設(shè)置數(shù)據(jù)value的反序列化處理類
    props.put("value.deserializer", StringDeserializer.class.getName());
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    KafkaConsumer kafkaConsumer = new KafkaConsumer<>(props);
    //訂閱名稱為“one-more-topic”的Topic的消息
    kafkaConsumer.subscribe(Arrays.asList("one-more-topic"));
    return kafkaConsumer;
  }
}

在Controller中進(jìn)行使用：

@RestController
@Slf4j
public class Controller {

  @Autowired
  private KafkaConsumer kafkaConsumer;

  @RequestMapping("/receive")
  public List receive() {
   從Kafka服務(wù)器中的名稱為“one-more-topic”的Topic中消費(fèi)消息
    ConsumerRecords records = kafkaConsumer.poll(Duration.ofSeconds(1));
    List messages = new ArrayList<>(records.count());
    for (ConsumerRecord record : records.records("one-more-topic")) {
      String message = record.value();
      log.info("message: {}", message);
      messages.add(message);
    }
    return messages;
  }
}

方式二：spring-kafka

使用kafka-clients需要我們自己創(chuàng)建生產(chǎn)者或者消費(fèi)者的bean，如果我們的項(xiàng)目基于SpringBoot構(gòu)建，那么使用spring-kafka就方便多了。

引入依賴

在pom.xml文件中，引入spring-kafka依賴：

  org.springframework.kafka
  spring-kafka
  2.3.12.RELEASE

生產(chǎn)者

在application.yml文件中增加配置：

spring:
 kafka:
  #Kafka服務(wù)器地址
  bootstrap-servers: 127.0.0.1:9092
  producer:
   #設(shè)置數(shù)據(jù)value的序列化處理類
   value-serializer: org.apache.kafka.common.serialization.StringSerializer

在Controller中注入KafkaTemplate就可以直接使用了，代碼如下：

@RestController
@Slf4j
public class Controller {

  @Autowired
  private KafkaTemplate template;

  @RequestMapping("/springKafkaSend")
  public String send() {
    String uuid = UUID.randomUUID().toString();
    //將消息發(fā)送到Kafka服務(wù)器的名稱為“one-more-topic”的Topic中
    this.template.send("one-more-topic", uuid);
    log.info("uuid: {}", uuid);
    return uuid;
  }
}

消費(fèi)者

在application.yml文件中增加配置：

spring:
 kafka:
  #Kafka服務(wù)器地址
  bootstrap-servers: 127.0.0.1:9092
  consumer:
   #設(shè)置數(shù)據(jù)value的反序列化處理類
   value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

創(chuàng)建一個(gè)可以被Spring框架掃描到的類，并且在方法上加上@KafkaListener注解，就可以消費(fèi)消息了，代碼如下：

@Component
@Slf4j
public class Receiver {

  @KafkaListener(topics = "one-more-topic", groupId = "spring-kafka-group")
  public void listen(ConsumerRecord record) {
    Optional kafkaMessage = Optional.ofNullable(record.value());
    if (kafkaMessage.isPresent()) {
      String message = (String) kafkaMessage.get();
      log.info("message: {}", message);
    }
  }
}

關(guān)于“如何使用Java實(shí)現(xiàn)Kafka生產(chǎn)者和消費(fèi)者的示例”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，使各位可以學(xué)到更多知識(shí)，如果覺(jué)得文章不錯(cuò)，請(qǐng)把它分享出去讓更多的人看到。

新聞標(biāo)題：如何使用Java實(shí)現(xiàn)Kafka生產(chǎn)者和消費(fèi)者的示例-創(chuàng)新互聯(lián)
網(wǎng)址分享：http://weahome.cn/article/cdhohi.html