快捷導(dǎo)航

Java實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者代碼實(shí)例

更新時(shí)間：2020年04月23日 09:26:26 作者：gdjlc

這篇文章主要介紹了Java實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

Kafka的結(jié)構(gòu)與RabbitMQ類似，消息生產(chǎn)者向Kafka服務(wù)器發(fā)送消息，Kafka接收消息后，再投遞給消費(fèi)者。
生產(chǎn)者的消費(fèi)會(huì)被發(fā)送到Topic中，Topic中保存著各類數(shù)據(jù)，每一條數(shù)據(jù)都使用鍵、值進(jìn)行保存。
每一個(gè)Topic中都包含一個(gè)或多個(gè)物理分區(qū)（Partition），分區(qū)維護(hù)著消息的內(nèi)容和索引，它們有可能被保存在不同服務(wù)器。

新建一個(gè)Maven項(xiàng)目，pom.xml 加入依賴：

    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>2.3.0</version>
    </dependency>

1、編寫生產(chǎn)者

將消息投遞到Kafka服務(wù)器的名稱為“topic1”的Topic中

package com.example.kafkatest;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服務(wù)器地址
    props.put("bootstrap.servers", "localhost:9092");
    //設(shè)置數(shù)據(jù)key和value的序列化處理類
    props.put("key.serializer", StringSerializer.class);
    props.put("value.serializer", StringSerializer.class);
    //創(chuàng)建生產(chǎn)者實(shí)例
    KafkaProducer<String,String> producer = new KafkaProducer<>(props);
    ProducerRecord record = new ProducerRecord<String, String>("topic1", "userName", "lc");
    //發(fā)送記錄
    producer.send(record);
    producer.close();
  }
}

運(yùn)行后，可打開命令行工具，進(jìn)入Kafka目錄，執(zhí)行命令查詢服務(wù)器的Topic：

bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

結(jié)果如下：

2、編寫消費(fèi)者

本例中，消費(fèi)者和生產(chǎn)者在同一個(gè)項(xiàng)目中，只是使用不同的啟動(dòng)類。

消費(fèi)者會(huì)為自已指定一個(gè)消費(fèi)者組的標(biāo)識(shí)，每一條發(fā)布到Topic的記錄，都會(huì)被交付給消費(fèi)者組的一個(gè)消費(fèi)者實(shí)例。

如果多個(gè)消費(fèi)者實(shí)例有相同的消費(fèi)者組，則這些記錄會(huì)分配到各個(gè)消費(fèi)者實(shí)例上，以達(dá)到負(fù)載均衡的目錄。

如果所有的消費(fèi)者有不同的消費(fèi)者組，則每一條記錄都會(huì)廣播到全部的消費(fèi)者進(jìn)行處理。

package com.example.rabbittest;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class Consumer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服務(wù)器地址
    props.put("bootstrap.servers", "localhost:9092");
    //必須指定消費(fèi)者組
    props.put("group.id", "test");
    //設(shè)置數(shù)據(jù)key和value的序列化處理類
    props.put("key.deserializer", StringDeserializer.class);
    props.put("value.deserializer", StringDeserializer.class);
    //創(chuàng)建消息者實(shí)例
    KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
    //訂閱topic1的消息
    consumer.subscribe(Arrays.asList("topic1"));
    //到服務(wù)器中讀取記錄
    while (true){
      ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
      for(ConsumerRecord<String,String> record : records){
        System.out.println("key:" + record.key() + "" + ",value:" + record.value());
      }
    }
  }
}

運(yùn)行后，IDEA控制臺(tái)其中輸出如下：