Java代码操作Kafka

1、消息的发送与接收

• 生产者主要的对象有： KafkaProducer ， ProducerRecord 。
• 其中KafkaProducer 是用于发送消息的类， ProducerRecord 类用于封装Kafka的消息。 

KafkaProducer 的创建需要指定的参数和含义：

其他参数可以从org.apache.kafka.clients.producer.ProducerConfig 中找到。

• 消费者生产消息后，需要broker端的确认，可以同步确认，也可以异步确认。
• 同步确认效率低，异步确认效率高，但是需要设置回调对象。

生产者：

package com.example.producer;
 
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
 
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
 
public class MyProducer1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map configs = new HashMap<>();
        // 指定初始连接用到的broker地址，如果是集群，则可以通过此初始连接发现集群中的其他broker
        configs.put("bootstrap.servers", "192.168.80.121:9092");
        // 指定key的序列化类
        configs.put("key.serializer", IntegerSerializer.class);
        // 指定value的序列化类
        configs.put("value.serializer", StringSerializer.class);
 
       /*  configs.put("acks", "all");
        configs.put("retries", 3); */
 
        // 创建kafkaProducer对象
        KafkaProducer producer = new KafkaProducer(configs);
 
        // 用于设置用户自定义的消息头字段
        List
 headers = new ArrayList<>();
        headers.add(new RecordHeader("biz.name", "producer.demo".getBytes()));
 
        // 封装消息
        ProducerRecord record = new ProducerRecord(
                "topic_1", // 主题名称
                0, // 分区编号，现在只有一个分区，所以是0
                0, // 数字作为key
                "hello world", // 字符串作为value
                headers
        );
 
        // 消息的同步确认
       /*  Future future = producer.send(record);
        RecordMetadata metadata = future.get();
        System.out.println("消息的主题：" + metadata.topic());
        System.out.println("消息的分区：" + metadata.partition());
        System.out.println("消息的偏移量：" + metadata.offset()); */
 
 
        // 消息的异步确认
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e==null) {
                    System.out.println("消息的主题：" + metadata.topic());
                    System.out.println("消息的分区：" + metadata.partition());
                    System.out.println("消息的偏移量：" + metadata.offset());
                }else{
                    System.out.println("异常消息:"+e.getMessage());
                }
            }
        });
 
        // 关闭生产者
        producer.close();
    }
}

消费者：

package com.example.consumer;
 
import org.apache.kafka.clients.consumer.ConsumerConfig;
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.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
 
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
 
public class MyConsumer {
    public static void main(String[] args) {
        Map configs = new HashMap<>();
        // 指定初始连接用到的broker地址
        // configs.put("bootstrap.servers", "192.168.80.121:9092");
        // 上面方式如果怕写错，可以尝试下面这种方法
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.80.121:9092");
        // 指定key的反序列化类
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        // 指定value的反序列化类
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 
        // 配置消费组id
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer_demo");
        // earliest:如果找不到当前消费者的有效偏移量，自自动重置到最开始
        // latest:表示直接重置到消息偏移量的最后一个
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 
        KafkaConsumer consumer = new KafkaConsumer(configs);
 
        // 先订阅，在消费
        consumer.subscribe(Arrays.asList("topic_1"));
 
      /*   while (true) {
            ConsumerRecords consumerRecords = consumer.poll(3_000);
        } */
        // 如果主题中没有可消费的消息，则该方法可以放到while循环中，每过3秒重新拉取一次
        // 如果还没有拉取到，过三秒再次拉取，防止while循环过于密集的poll调用
 
        // 批量从主题的分区拉取消息
        ConsumerRecords consumerRecords = consumer.poll(3_000); // 指定拉取消息的时间间隔
 
        // 遍历本次从主题分区拉取的批量消息
        consumerRecords.forEach(new Consumer>() {
            @Override
            public void accept(ConsumerRecord record) {
                System.out.println("========================================");
                System.out.println("消息头字段：" + Arrays.toString(record.headers().toArray()));
                System.out.println("消息的key：" + record.key());
                System.out.println("消息的偏移量：" + record.offset());
                System.out.println("消息的分区号：" + record.partition());
                System.out.println("消息的序列化key字节数：" + record.serializedKeySize());
                System.out.println("消息的序列化value字节数：" + record.serializedValueSize());
                System.out.println("消息的时间戳：" + record.timestamp());
                System.out.println("消息的时间戳类型：" + record.timestampType());
                System.out.println("消息的主题：" + record.topic());
                System.out.println("消息的值：" + record.value());
            }
        });
 
        consumer.close();
    }
}

2、SpringBoot Kafka

（1）pom.xml文件

"1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0modelVersion>
    <parent>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-parentartifactId>
        <version>2.2.8.RELEASEversion>
        <relativePath/> 
    parent>
    <groupId>com.examplegroupId>
    <artifactId>demo_02_springboot-kafkaartifactId>
    <version>0.0.1-SNAPSHOTversion>
    <name>demo_02_springboot-kafkaname>
    <description>demo_02_springboot-kafkadescription>
    <properties>
        <java.version>8java.version>
    properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
        dependency>
        <dependency>
            <groupId>org.springframework.kafkagroupId>
            <artifactId>spring-kafkaartifactId>
        dependency>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-testartifactId>
            <scope>testscope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintagegroupId>
                    <artifactId>junit-vintage-engineartifactId>
                exclusion>
            exclusions>
        dependency>
        <dependency>
            <groupId>org.springframework.kafkagroupId>
            <artifactId>spring-kafka-testartifactId>
            <scope>testscope>
        dependency>
    dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.bootgroupId>
                <artifactId>spring-boot-maven-pluginartifactId>
            plugin>
        plugins>
    build>
 
project>

（2）application.properties

spring.application.name=springboot-kafka-02
server.port=8080
 
# 用于建立初始连接的broker地址
spring.kafka.bootstrap-servers=192.168.80.121:9092
 
# producer用到的key和value的序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.IntegerSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 
# 默认的批处理记录数
spring.kafka.producer.batch-size=16384
# 32MB的总发送缓存
spring.kafka.producer.buffer-memory=33554432
 
# consumer用到的key和value的反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# consumer的消费组id
spring.kafka.consumer.group-id=springboot-consumer02
 
# 是否自动提交消费者偏移量
spring.kafka.consumer.enable-auto-commit=true
# 每隔100ms向broker提交一次偏移量
spring.kafka.consumer.auto-commit-interval=100
# 如果该消费者的偏移量不存在，则自动设置为最早的偏移量
spring.kafka.consumer.auto-offset-reset=earliest

（3）启动类

package com.example.demo;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class Demo02SpringbootKafkaApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(Demo02SpringbootKafkaApplication.class, args);
    }
 
}

（4）KafkaConfig（在这里可以进行主题的创建、自定义了kafkaAdmin对象等一系列配置，也可以省略，如果kafka在连接主题时，发现没有，KafkaAdmin这个类会自动帮我们创建）

package com.example.demo.config;
 
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class KafkaConfig {
 
    // 创建主题
    @Bean
    public NewTopic topic1() {
        return new NewTopic("nptc-01",3, (short) 1);
    }
 
    @Bean
    public NewTopic topic2() {
        return new NewTopic("nptc-02",5, (short) 1);
    }
 
 
}

（5）生产者

同步方式 

package com.example.demo.controlller;
 
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
 
@RestController
public class KafkaSyncProducerController {
    @Autowired
    private KafkaTemplate template;
 
    @RequestMapping("send/sync/{message}")
    public String send(@PathVariable String message) throws ExecutionException, InterruptedException {
        ListenableFuture> future = template.send("topic-spring-01", 0, 0, message);
 
        // 同步发送消息
        SendResult sendResult = future.get();
        RecordMetadata metadata = sendResult.getRecordMetadata();
        System.out.println(metadata.topic() + "\t" + metadata.partition() + "\t" + metadata.offset());
        return "success";
    }
 
}

异步回调方式

package com.example.demo.controlller;
 
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
@RestController
public class KafkaAsyncProducerController {
    @Autowired
    private KafkaTemplate template;
 
    @RequestMapping("send/async/{message}")
    public String send(@PathVariable String message) {
        ListenableFuture> future = (ListenableFuture>) template.send("topic-spring-01", 0, 1, message);
 
        future.addCallback(new ListenableFutureCallback>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败：" + ex.getMessage());
            }
 
            @Override
            public void onSuccess(SendResult result) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("发送消息成功：" + metadata.topic() + "\t" + metadata.partition() + "\t" + metadata.offset());
            }
        });
 
        return "success";
    }
 
}

（6）消费者

package com.example.demo.consumer;
 
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class MyConsumer {
    @KafkaListener(topics = "topic-spring-01")
    public void onMessage(ConsumerRecord record) {
        System.out.println("消费者收到的消息：" + record.topic() + "\t" + record.partition() + "\t" + record.offset() + "\t" + record.key() + "\t" + record.value());
    }
}