当前位置： 首页 > news >正文

从零开始的 Kafka 学习（四）| 生产消息

news 来源：原创 2025/6/9 0:22:55

1. 生产消息

1.1 生产消息的基本步骤

（一）创建Map类型的配置对象，根据场景增加相应的配置属性：

参数名参数作用类型默认值推荐值
bootstrap.servers集群地址，格式为：brokerIP1:端口号,brokerIP2:端口号必须
key.serializer对生产数据Key进行序列化的类完整名称必须Kafka提供的字符串序列化类：StringSerializer
value.serializer对生产数据Value进行序列化的类完整名称必须Kafka提供的字符串序列化类：StringSerializer
interceptor.classes拦截器类名，多个用逗号隔开可选
batch.size数据批次字节大小。此大小会和数据最大估计值进行比较，取大值。估值=61+21+（keySize+1+valueSize+1+1）可选16K
retries重试次数可选整型最大值0或整型最大值
request.timeout.ms请求超时时间可选30s
linger.ms数据批次在缓冲区中停留时间可选
acks请求应答类型：all(-1), 0, 1可选all(-1)根据数据场景进行设置
retry.backoff.ms两次重试之间的时间间隔可选100ms
buffer.memory数据收集器缓冲区内存大小可选32M64M
max.in.flight.requests.per.connection每个节点连接的最大同时处理请求的数量可选5小于等于5
enable.idempotence幂等性，可选true根据数据场景进行设置
partitioner.ignore.keys是否放弃使用数据key选择分区可选false
partitioner.class分区器类名可选null

（二）创建待发送数据

在 Kafka 中传递的数据我们称之为消息（message）或记录（record），所以Kafka发送数据前，需要将待发送的数据封装为指定的数据类型：

在这里插入图片描述

在这里插入图片描述

相关属性必须在构建数据模型时指定，其中主题和value的值时必须要传递的。如果配置中开启了自动创建主题，那么 Topic 主题可以不存在。value 就是我们需要真正传递的数据了，而 Key 可以用于数据的分区定位。

（三）创建生产者对象，发送生产的数据：

根据前面提供的配置信息创建生产者对象，通过这个生产者对象向 Kafka 服务器节点发送数据，而具体的发送是由生产者对象创建时，内部构件的多个组件实现的，多个组件的关系类似与生产者消费者模式。

在这里插入图片描述

（1）数据生产者（KafkaProducer）：生产者对象，用于对我们的数据进行必要的转换和处理，将处理后的数据放入到数据收集器中，类似于生产者消费者模式下的生产者。

  • 如果配置拦截器栈（interceptor.classes），那么将数据进行拦截处理。某一个拦截器出现异常并不会影响后续的拦截器处理。
  • 因为发送的数据为 KV 数据，所以需要根据配置信息中的序列化对象对数据中 Key 和 Value 分别进行序列化处理。
  • 计算数据嗦发送的分区位置。
  • 将数据追加到数据收集器中。

（2）数据收集器（RecordAccumulator）：用于收集，转换我们生产的数据，蕾西与生产者消费者模式下的缓冲区。为了优化数据的传输，Kafka 并不是生产一条数据就向 Broker 发送一条数据，而是通过合并单条消息，进行批量（批次）发送，提高吞吐量，减少带宽消耗。

  • 默认情况下，一个发送批次的数据容量为 16k，这个可以通过参数 batch.size进行改善。
  • 批次是和分区进行绑定的。也就是说发往同一个分区的数据会进行合并，形成一个批次。
  • 如果当前批次能容纳数据，那么直接将数据追加到批次中即可，如果不能容纳数据，那么会产生新的批次放入到当前分区的批次队列中，这个队列使用的是 Java 双端队列 Deque。旧的批次关闭不再接收新的数据，等待发送。

（3）数据发送器（Sender）：线程对象，用于从收集器中获取数据，向服务节点发送。类似于生产者消费者模式下的消费者。因为是线程对象，所以启动后会不断轮询获取数据收集器中已经关闭的批次数据。对批次进行整合后再发送到 Broker 节点中

  • 因为数据真正发送的地方是 Broker 节点，不是分区。所以需要将从数据收集器中收集到的批次数据按照可用 Broker 节点重新组合成List集合。
  • 将组合后的<节点，List<批次>>的数据封装成客户端请求（请求键为：Produce）发送到网络客户端对象的缓冲区，由网络客户端对象通过网络发送给 Broker 节点。
  • Broker 节点获取客户端请求，并根据请求键进行后续的数据处理：向分区中增加数据。

在这里插入图片描述

1.2 生产消息的基本代码

// TODO 配置属性集合
Map<String, Object> configMap = new HashMap<>();
// TODO 配置属性：Kafka服务器集群地址
configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// TODO 配置属性：Kafka生产的数据为KV对，所以在生产数据进行传输前需要分别对K,V进行对应的序列化操作
configMap.put(
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
configMap.put(
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
// TODO 创建Kafka生产者对象，建立Kafka连接
//      构造对象时，需要传递配置参数
KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);
// TODO 准备数据,定义泛型
//      构造对象时需要传递 【Topic主题名称】，【Key】，【Value】三个参数
ProducerRecord<String, String> record = new ProducerRecord<String, String>(
        "test", "key1", "value1"
);
// TODO 生产（发送）数据
producer.send(record);
// TODO 关闭生产者连接
producer.close();

1.3 发送消息

1.3.1 拦截器

生产者 API 在数据准备好发送给 Kafka 服务器之前，允许我们对生产的数据进行统一的处理，比如校验，整合数据等等。这些处理我们是可以通过 Kafka 提供的拦截器完成。

这里的拦截器是可以配置多个的。执行时，会按照声明顺序执行完一个后，再执行下一个。并且某一个拦截器如果出现异常，只会跳出当前拦截器逻辑，并不会影响后续拦截器的处理。所以开发时，需要将拦截器的这种处理方法考虑进去。

在这里插入图片描述

1.3.1.1 增加拦截器类

（1）实现生产者拦截器接口 ProducerInterceptor

/**
 * TODO 自定义数据拦截器
 *      1. 实现Kafka提供的生产者接口ProducerInterceptor
 *      2. 定义数据泛型 <K, V>
 *      3. 重写方法
 *         onSend
 *         onAcknowledgement
 *         close
 *         configure
 */
public class KafkaInterceptorMock implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return record;
    }

    @Override	
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}

（2）实现接口中的方法，根据业务功能重写具体的方法

方法名作用
onSend数据发送前，会执行此方法，进行数据发送前的预处理
onAcknowledgement数据发送后，获取应答时，会执行此方法
close生产者关闭时，会执行此方法，完成一些资源回收和释放的操作
configure创建生产者对象的时候，会执行此方法，可以根据场景对生产者对象的配置进行统一修改或转换。
1.3.1.2 配置拦截器
public class ProducerInterceptorTest {
    public static void main(String[] args) {
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configMap.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        configMap.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        configMap.put( ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaInterceptorMock.class.getName());

        KafkaProducer<String, String> producer = null;
        try {
            producer = new KafkaProducer<>(configMap);
            for ( int i = 0; i < 1; i++ ) {
                ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", "key" + i, "value" + i);
                final Future<RecordMetadata> send = producer.send(record);
            }
        } catch ( Exception e ) {
            e.printStackTrace();
        } finally {
            if ( producer != null ) {
                producer.close();
            }
        }

    }
}
1.3.2 回调方法

Kafka 发送数据时，可以同时传递回调对象（Callback）用于对数据的发送结果进行对应处理，具体代码实现采用匿名类或 Lambda 表达式都可以。

public class KafkaProducerASynTest {
    public static void main(String[] args) {
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configMap.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);
        // TODO 循环生产数据
        for ( int i = 0; i < 1; i++ ) {
            // TODO 创建数据
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", "key" + i, "value" + i);
            // TODO 发送数据
            producer.send(record, new Callback() {
                // TODO 回调对象
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // TODO 当数据发送成功后，会回调此方法
                    System.out.println("数据发送成功：" + recordMetadata.timestamp());
                }
            });
        }
        producer.close();
    }
}
1.3.3 异步发送

Kafka 发送数据时，底层的实现类似于生产者消费者模式。对应的，底层会由主线程代码作为生产者向缓冲区中放数据，而数据发送线程会从缓冲区中获取数据进行发送。Broker 接收到数据后进行后续处理。

如果 Kafka 通过主线程代码将一条数据放入到缓冲区后，无需等待数据的后续发送过程，就直接发送下一条数据的场合，我们就称之为异步发送。

在这里插入图片描述

public class KafkaProducerASynTest {
    public static void main(String[] args) {
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configMap.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);
        // TODO 循环生产数据
        for ( int i = 0; i < 10; i++ ) {
            // TODO 创建数据
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", "key" + i, "value" + i);
            // TODO 发送数据
            producer.send(record, new Callback() {
                // TODO 回调对象
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // TODO 当数据发送成功后，会回调此方法
                    System.out.println("数据发送成功：" + recordMetadata.timestamp());
                }
            });
            // TODO 发送当前数据
            System.out.println("发送数据");
        }
        producer.close();
    }
}
1.3.4 同步发送

Kafka 发送数据时，底层的实现类似于生产者消费者模式。对应的，底层会由主线程代码作为生产者向缓冲区中放数据，而数据发送线程会从缓冲区中获取数据进行发送。Broker 接收到数据后进行后续处理。

如果 Kafka 通过主线程代码将一条数据放入到缓冲区后，需等待数据的后续发送操作的应答状态，才能发送下一条数据的场合，我们就称之为同步发送。所以这里的所谓同步，就是生产数据的线程需要等待线程的应答（响应）结果。

代码实现上，采用的是 JDK1.5 增加的JUC 并发编程的 Future 接口的 get 方法实现。

在这里插入图片描述

public class KafkaProducerASynTest {
    public static void main(String[] args) throws Exception {
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configMap.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);
        // TODO 循环生产数据
        for ( int i = 0; i < 10; i++ ) {
            // TODO 创建数据
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", "key" + i, "value" + i);
            // TODO 发送数据
            producer.send(record, new Callback() {
                // TODO 回调对象
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // TODO 当数据发送成功后，会回调此方法
                    System.out.println("数据发送成功：" + recordMetadata.timestamp());
                }
            }).get();
            // TODO 发送当前数据
            System.out.println("发送数据");
        }
        producer.close();
    }
}

相关文章：

  • 【python】requests 爬虫高效获取游戏皮肤图
  • 文件上传绕过的小点总结(4)
  • ubuntu22.04 安装Jitsi meet 开源会议系统，代替腾讯会议
  • 用fofa语法搜索漏洞
  • JMeter JSON断言讲解和错误用例
  • JAVA学习*String类
  • SpringBoot集成MQTT客户端
  • 16种二极管综合对比——《器件手册--二极管》
  • 【第十三届“泰迪杯”数据挖掘挑战赛】【2025泰迪杯】A题解题全流程（持续更新）
  • #MySQL数据库深度瘦身优化技术方案
  • linux常用指令（7）
  • 【Infra】理解代理、反向代理和负载均衡器
  • 【Springboot+JPA】存储过程Procedure使用及踩坑记录
  • Spring 源码硬核解析系列专题（三十二）：Spring Cloud LoadBalancer 的负载均衡源码解析
  • [杂学笔记]结构体大小的内存对齐规则、继承与组合的区别、New和malloc的区别、define与const的区别、如何保证线程安全、乐观锁与悲观锁
  • 状态空间模型解析 (State-Space Model, SS）
  • Vue 2 探秘：visible 和 append-to-body 是谁的小秘密？
  • 低代码配置式Web组态解析
  • 运维知识：Linux 密码策略：配置、注意事项与实战详解
  • 新手小白 react-useEffect 使用场景
  • 报名入口网站建设/免费发外链平台
  • 怎么做网站竞价推广/淘宝seo什么意思
  • 网站和域名的关系/关键词挖掘站长
  • php做网站需要html/杭州百度seo优化
  • 大学网站建设专业/网络营销的特点分别是
  • 十堰有做百度推广的吗/邯郸seo优化公司