Kafka3.0.0版本——生产者如何提高吞吐量

这篇具有很好参考价值的文章主要介绍了Kafka3.0.0版本——生产者如何提高吞吐量。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、生产者如何提高吞吐量

  • 设置批次大小,batch.size 默认 16K。
  • 设置等待时间,linger.ms 默认 0。
  • 设置缓冲区大小,buffer.memory 默认 32M
  • 设置压缩, compression.type 默认 none,可配置值 gzip、snappy、lz4 和 zstd。

二、生产者提高吞吐量代码示例

  • 代码

    package com.xz.kafka.producer;
    
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.StringSerializer;
    import java.util.Properties;
    /**
     * 生产者提高吞吐量
     * 1、设置批次大小,batch.size 默认 16K
     * 2、设置等待时间,linger.ms 默认 0
     * 3、设置缓冲区大小,buffer.memory 默认 32M
     * 4、设置压缩, compression.type 默认 none,可配置值 gzip、snappy、lz4 和 zstd
     * */
    public class CustomProducerHandlingCapacity {
    
        public static void main(String[] args) throws InterruptedException {
    
            //1、创建 kafka 生产者的配置对象
            Properties properties = new Properties();
    
            //2、给 kafka 配置对象添加配置信息:bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");
    
            //3、指定对应的key和value的序列化类型 key.serializer value.serializer
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    
            /**
             * 4、提高吞吐量
             * */
            //设置缓冲区大小
            properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
            //设置批次大小
            properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
            //设置等待时间 linger.ms
            properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
            //设置压缩
            properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
    
    
            //5、创建生产者
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
    
            //6、调用 send 方法,发送消息
            for (int i = 0; i < 5; i++) {
                kafkaProducer.send(new ProducerRecord<>("news", "hello kafka" + i), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null){
                            System.out.println("主题: "+metadata.topic() + " 分区: "+ metadata.partition());
                        }
                    }
                });
                Thread.sleep(2);
            }
    
            //7、关闭资源
            kafkaProducer.close();
        }
    }
    
  • 在kafka集群上开启 Kafka 消费者

    [root@localhost kafka-3.0.0]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.136.27:9092 --topic news
    
  • 在 IDEA 中执行代码,观察kafka集群控制台中是否接收到消息。
    Kafka3.0.0版本——生产者如何提高吞吐量

  • 测试吞吐量需要大数据量,因此lz不做大数据量示例演示。文章来源地址https://www.toymoban.com/news/detail-409111.html

到了这里,关于Kafka3.0.0版本——生产者如何提高吞吐量的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • [kafka消息生产被阻塞] - 如何解决Kafka生产者阻塞的问题

    [kafka消息生产被阻塞] - 如何解决Kafka生产者阻塞的问题 Kafka是一个高度可扩展的分布式流平台,用于构建实时数据管道和流处理应用程序。作为一个广泛使用的消息代理系统,Kafka在数据传输方面表现出色,但是在极端情况下,它可能会出现生产者阻塞的问题。这可能会导致

    2024年02月11日
    浏览(46)
  • 三、Kafka生产者1---Kafka生产者初始化-new KafkaProducer

    概述 本文主要是分享Kafka初始化生产者的 大体过程 初始化过程中会新建很多对象,本文暂先分享部分对象 1.分区器---Partitioner partitioner 2.重试时间---long retryBackoffMs 3.序列化器---SerializerK keySerializer,SerializerV valueSerializer 4.拦截器--- ListProducerInterceptorK, V interceptorList 5.累加器-

    2024年03月14日
    浏览(60)
  • kafka学习-生产者

    目录 1、消息生产流程 2、生产者常见参数配置 3、序列化器 基本概念 自定义序列化器 4、分区器 默认分区规则 自定义分区器 5、生产者拦截器 作用 自定义拦截器 6、生产者原理解析 在Kafka中保存的数据都是字节数组。 消息发送前,需要将消息序列化为字节数组进行发送。

    2024年02月09日
    浏览(39)
  • (三)Kafka 生产者

    创建一个 ProducerRecord 对象,需要包含目标主题和要发送的内容,还可以指定键、分区、时间戳或标头。 在发送 ProducerRecord 对象时,生产者需要先把键和值对象序列化成字节数组,这样才能在网络上传输。 如果没有显式地指定分区,那么数据将被传给分区器。分区器通常会基

    2024年02月09日
    浏览(38)
  • Kafka 生产者

    目录 一、kafka生产者原理 二、kafka异步发送 配置kafka 创建对象,发送数据 带回调函数的异步发送 同步发送   三、kafka生产者分区 分区策略 指定分区:  指定key: 什么都不指定: 自定义分区器 四、生产者提高吞吐量 五、数据的可靠性 ACK应答级别 数据完全可靠条件 可靠性

    2023年04月15日
    浏览(43)
  • 「Kafka」生产者篇

    在消息发送的过程中,涉及到了 两个线程 —— main 线程 和 Sender 线程 。 在 main 线程中创建了 一个 双端队列 RecordAccumulator 。 main线程将消息发送给RecordAccumulator,Sender线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。 main线程创建 Producer 对象,调用 send 函数发送消息,

    2024年01月19日
    浏览(39)
  • Kafka-生产者

    Kafka在实际应用中,经常被用作高性能、可扩展的消息中间件。 Kafka自定义了一套网络协议,只要遵守这套协议的格式,就可以向Kafka发送消息,也可以从Kafka中拉取消息。 在实践生产过程中,一套API封装良好、灵活易用的客户端可以避免开发人员重复劳动,提高开发效率,也

    2024年01月20日
    浏览(34)
  • Kafka(生产者)

    目 前 企 业 中 比 较 常 见 的 消 息 队 列 产 品 主 要 有 Kafka(在大数据场景主要采用 Kafka 作为消息队列。) ActiveMQ RabbitMQ RocketMQ 1.1.1 传统消息队列的应用场景 传统的消息队列的主要应用场景包括: 缓存/消峰 、 解耦 和 异步通信 。 缓冲/消峰: 有助于控制和优化数据流经过

    2024年02月11日
    浏览(44)
  • 三、Kafka生产者

    1 发送原理 在消息发送的过程中,涉及到了两个线程——main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator,Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker 【RecordAccumulator缓冲的结构: 每一个分区对应一

    2024年02月12日
    浏览(37)
  • Kafka生产者

    1.acks 如果acks=0,生产者在成功写入消息之前不会等待任何来自服务器的响应。 缺点:如果当中出现了问题,导致服务器没有收到消息,那么生产者就无从得知,消息就丢失了 优点:因为生产者不需要等待服务器的响应,所有他可以以网络能够支持的最大速度发送消息,从而

    2024年01月19日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包