前言

本文主要是使用 Java 语言中 spring-kafka 依赖 对 Kafka 进行使用。

使用以下依赖对 Kafka 进行操作：

<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.11</version>
</dependency>

需要更改版本的话，可以前往：Maven 仓库

创建项目，先创建一个简单的 Maven 项目，删除无用的包、类之后，使用其作为一个父级项目。
以下内容如果在项目启动时报这个错：

org.yaml.snakeyaml.error.YAMLException: java.nio.charset.MalformedInputException: Input length = 1

把注释删除就可以了。

写法一：发送的消息对象是字符串

1 创建项目

随后创建SpringBoot模块。选择 Kafka 组件

随后调整该项目的POM依赖为：

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>11</source>
                    <target>11</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

注意这一步同时需要将Java版本、Maven版本都调整好。我这里目前使用的是Java11。

2 项目结构

我们先看 spring-kafka-demo模块的内容

3 application.yml 配置文件

主要指定集群信息、生产者信息、消费者信息。尤其重要的是序列化方式。

server:
  # 优雅停机
  shutdown: graceful

spring:
  kafka:
    # kafka集群信息，多个用逗号间隔
    bootstrap-servers: localhost:9092
    # 生产者
    producer:
      # 重试次数，设置大于0的值，则客户端会将发送失败的记录重新发送
      retries: 3
      #批量处理大小，16K
      batch-size: 16384
      #缓冲存储大，32M
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息体的编码方式：字符串序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # 消费者
    consumer:
      # 消费者组
      group-id: TestGroup
      # 是否自动提交
      enable-auto-commit: false
      # 消费偏移配置
      # none：如果没有为消费者找到先前的offset的值,即没有自动维护偏移量,也没有手动维护偏移量,则抛出异常
      # earliest：在各分区下有提交的offset时：从offset处开始消费；在各分区下无提交的offset时：从头开始消费
      # latest：在各分区下有提交的offset时：从offset处开始消费；在各分区下无提交的offset时：从最新的数据开始消费
      auto-offset-reset: latest
      # 指定消息key和消息体的解码方式：字符串反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    # 监听
    listener:
      # record：当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
      # batch：当每一批poll()的数据被ListenerConsumer处理之后提交
      # time：当每一批poll()的数据被ListenerConsumer处理之后，距离上次提交时间大于TIME时提交
      # count：当每一批poll()的数据被ListenerConsumer处理之后，被处理record数量大于等于COUNT时提交
      # count_time：TIME或COUNT中有一个条件满足时提交
      # manual：当每一批poll()的数据被ListenerConsumer处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # manual_immediate：手动调用Acknowledgment.acknowledge()后立即提交，一般推荐使用这种
      ack-mode: manual_immediate

4 生产者 KafkaProducerComponent

package org.feng.kafka.sender;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * Kafka消息生产者组件
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月16日 23时26分
 */
@Slf4j
@Component
public class KafkaProducerComponent {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 预先在 Kafka 中创建好的 topic
     */
    public static final String TOPIC = "testTopic";

    public void send(String topic, String data) {
        kafkaTemplate.send(topic, data)
                // 回调
                .addCallback(new ListenableFutureCallback<>() {
                    @Override
                    public void onFailure(@NonNull Throwable throwable) {
                        log.error("主题[{}]发送消息[{}]失败", topic, data, throwable);
                    }

                    @Override
                    public void onSuccess(SendResult<String, String> result) {
                        log.info("主题[{}]发送消息[{}]成功", topic, data);
                    }
                });
    }
}

5 消费者 KafkaConsumerComponent

package org.feng.kafka.receiver;

import lombok.extern.slf4j.Slf4j;
import org.feng.kafka.sender.KafkaProducerComponent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 监听消息：消费端
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 19时54分
 */
@Slf4j
@Component
public class KafkaConsumerComponent {

    @KafkaListener(topics = KafkaProducerComponent.TOPIC)
    public void consumerTestTopic(String data) {
        log.info("消费者监听到数据：{}", data);
    }
}

6 控制器（GET请求发送消息）

package org.feng.kafka.controller;

import lombok.extern.slf4j.Slf4j;
import org.feng.kafka.sender.KafkaProducerComponent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * 发送消息控制器
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 19时56分
 */
@Slf4j
@RestController
public class SendMessageController {

    @Resource
    private KafkaProducerComponent kafkaProducerComponent;

    @GetMapping("/send/{data}")
    public String send(@PathVariable("data") String data) {
        log.info("即将把数据【{}】发送到消息队列", data);
        kafkaProducerComponent.send(KafkaProducerComponent.TOPIC, data);
        return "send ok";
    }
}

7 启动类

package org.feng.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringKafkaDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringKafkaDemoApplication.class, args);
    }
}

8 测试效果

我这边已经启动了 Kafka ，随后在本地再启动本项目，待项目启动后，使用 GET 请求给 Kafka 中扔消息。

使用以上的链接触发。
可以依次观察到日志记录：

即将把数据【测试呢111】发送到消息队列
消费者监听到数据：测试呢111
主题[testTopic]发送消息[测试呢111]成功

写法二：发送复杂消息对象

其实就是自定义一个对象，直接扔到消息队列里。然后再使用监听器监听到，并作出处理。
核心改变的地方是消息Value 的序列化方式、反序列化方式，更改为：

修改了 group-id、值的序列化、反序列化，以及增加了属性“信任的包”。
你想把哪个类的对象放消息队列，就得在这个包下进行定义这个类。

1 创建项目

项目的版本和写法一保持一致。
包括 POM 文件也是一致的。

2 项目结构

3 application.yml 配置文件

PS：这里将注释几乎全部去掉了

server:
  shutdown: graceful
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      retries: 3
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值序列化：使用Json
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: TestObjectGroup
      enable-auto-commit: false
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 值反序列化：使用Json
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      # 信任的包
      properties:
        spring:
          json:
            trusted:
              packages: org.feng.entity
    listener:
      ack-mode: manual_immediate

4 信任的包中定义的实体类

4.1 kafka 消息接口规则定义

package org.feng.entity;

/**
 * kafka 消息
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 20时16分
 */
public interface KafkaMessage {
}

4.2 测试实体定义

实体实现了 KafkaMessage 规则。
并定义了简单的属性值。

package org.feng.entity;

import lombok.Data;

import java.time.LocalDateTime;
import java.util.Locale;
import java.util.UUID;

/**
 * 测试kafka消息对象
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 20时18分
 */
@Data
public class TestKafkaMessage implements KafkaMessage {
    private LocalDateTime time = LocalDateTime.now();

    private String message;

    private String business = "test";

    private String messageId = UUID.randomUUID().toString().toLowerCase(Locale.ROOT).replaceAll("-", "");
}

5 生产者 KafkaObjectSerializerProducerComponent

package org.feng.producer;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.feng.entity.KafkaMessage;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * 生产者
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 20时21分
 */
@Slf4j
@Component
public class KafkaObjectSerializerProducerComponent {
    /**
     * 预先在 Kafka 中创建好的 topic
     */
    public static final String TOPIC = "testObjectTopic";

    @Resource
    private KafkaTemplate<String, ? super KafkaMessage> kafkaTemplate;

    public void sendTest(String topic, KafkaMessage kafkaMessage) {
        kafkaTemplate.send(topic, kafkaMessage)
                // 回调
                .addCallback(new ListenableFutureCallback<SendResult<String, ? super KafkaMessage>>() {
                    @Override
                    public void onFailure(@NonNull Throwable throwable) {
                        log.error("主题[{}]发送消息[{}]失败", topic, kafkaMessage, throwable);
                    }

                    @Override
                    public void onSuccess(SendResult<String, ? super KafkaMessage> result) {
                        log.info("主题[{}]发送消息[{}]成功，发送结果：{}", topic, kafkaMessage, result);
                    }
                });
    }
}

6 消费者 KafkaObjectSerializerConsumerComponent

package org.feng.consumer;

import lombok.extern.slf4j.Slf4j;
import org.feng.entity.TestKafkaMessage;
import org.feng.producer.KafkaObjectSerializerProducerComponent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 消费者
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 20时30分
 */
@Component
@Slf4j
public class KafkaObjectSerializerConsumerComponent {

    @KafkaListener(topics = KafkaObjectSerializerProducerComponent.TOPIC)
    public void consumerTestTopic(TestKafkaMessage data) {
        log.info("消费者监听到数据：{}", data);
    }
}

7 控制器（GET请求发送消息）

重点在于，消息内容是自定义的 TestKafkaMessage 实例。

package org.feng.controller;

import lombok.extern.slf4j.Slf4j;
import org.feng.entity.TestKafkaMessage;
import org.feng.producer.KafkaObjectSerializerProducerComponent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * 发送消息控制器
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 19时56分
 */
@Slf4j
@RestController
public class SendMessageController {

    @Resource
    private KafkaObjectSerializerProducerComponent kafkaObjectSerializerProducerComponent;

    @GetMapping("/send")
    public String send(@RequestBody TestKafkaMessage data) {
        log.info("即将把数据【{}】发送到消息队列", data);
        kafkaObjectSerializerProducerComponent.sendTest(KafkaObjectSerializerProducerComponent.TOPIC, data);
        return "send ok";
    }
}

8 启动类

和写法一基本一致（除了类名不同）

9 测试效果

我这边已经启动了 Kafka ，随后在本地再启动本项目，待项目启动后，使用 GET 请求给 Kafka 中扔消息。

使用以上的链接触发。
可以依次观察到日志记录：

即将把数据【TestKafkaMessage(time=2023-03-17T21:29:21.564849900, message=消息内容就是我了, business=test, messageId=aa12c2e29bed431090918d971477de16)】发送到消息队列
消费者监听到数据：TestKafkaMessage(time=2023-03-17T21:29:21.564849900, message=消息内容就是我了, business=test, messageId=aa12c2e29bed431090918d971477de16)
主题[testObjectTopic]发送消息[TestKafkaMessage(time=2023-03-17T21:29:21.564849900, message=消息内容就是我了, business=test, messageId=aa12c2e29bed431090918d971477de16)]成功，发送结果：SendResult [producerRecord=ProducerRecord(topic=testObjectTopic, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [111, 114, 103, 46, 102, 101, 110, 103, 46, 101, 110, 116, 105, 116, 121, 46, 84, 101, 115, 116, 75, 97, 102, 107, 97, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=TestKafkaMessage(time=2023-03-17T21:29:21.564849900, message=消息内容就是我了, business=test, messageId=aa12c2e29bed431090918d971477de16), timestamp=null), recordMetadata=testObjectTopic-0@1]

附录

1 减少日志输出

默认情况下，Kafka的日志很多都会打印出来，但是又与我们业务本身无关。需要屏蔽一下。
这里做了简单的处理，使用 logback 设置了日志级别。

logback 文件内容如下：

<?xml version="1.0" encoding="UTF-8" ?>
<configuration debug="false">
    <!-- 配置控制台输出 -->
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!-- 格式化输出： %d表示日期， %thread表示线程名， %-5level: 级别从左显示5个字符宽度 %msg:日志消息, %n是换行符 -->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS}[%thread] %-5level %logger{50} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 日志输出级别 -->
    <root level="INFO">
        <appender-ref ref="STDOUT"/>
    </root>
    <!-- 定制化某些包的日志输出级别 -->
    <logger name="org.apache.kafka" level="warn" additivity="false"/>
    <logger name="org.springframework" level="info">
        <appender-ref ref="STDOUT" />
    </logger>
</configuration>

效果如下：

发现日志确实少了很多。这样也方便我们后续开发。

2 手动提交偏移量

细心的朋友们可能已经发现了，以上的实例中，在项目重新启动时，会自动消费几条数据，这主要是因为我们设置了“不自动提交偏移量”，但是程序中又没有去手动提交。
现在我们来处理这个问题，首先是对原先的配置进行微调：

server:
  shutdown: graceful
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      retries: 3
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: TestObjectGroup
      # 依然使用非自动提交
      enable-auto-commit: false
      # 修改读取的偏移量的方式
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            trusted:
              packages: org.feng.entity
    listener:
      ack-mode: manual
      # 设置并发量
      concurrency: 3

以上修改了读取偏移量的方式为：在各分区下有提交的offset时，从offset处开始消费；在各分区下无提交的offset时：从头开始消费
然后调整监听者的配置 ack-mode: manual，当每一批poll()的数据被消费端处理之后, 手动调用Acknowledgment.acknowledge()后提交。
监听器的写法上也做一下调整：

package org.feng.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.feng.entity.TestKafkaMessage;
import org.feng.producer.KafkaObjectSerializerProducerComponent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 消费者
 *
 * @version v1.0
 * @author: fengjinsong
 * @date: 2023年03月17日 20时30分
 */
@Component
@Slf4j
public class KafkaObjectSerializerConsumerComponent {
    
    @KafkaListener(topics = KafkaObjectSerializerProducerComponent.TOPIC)
    public void consumeTestTopicAndCommit(ConsumerRecord<String, TestKafkaMessage> record, Acknowledgment ack) {
        try {
            log.info("消费者监听到数据：{}", record.value());
            // 手动提交
            ack.acknowledge();
        } catch (Exception e) {
            log.info("消费失败，数据：{}", record.value(), e);
        }
    }
}

监听器的主要调整在于方法入参、消费者处理消息后增加手动提交的操作。