在apache kafka生产者方压缩
我在producer端启用了snappy压缩，批量大小为64kb，处理每个1KB的消息，并将延迟时间设置为inf，这是否意味着在我处理64条消息之前，producer不会将消息发送给kafka out topic。。。Kafk64将发送一条消息，或每一个消息的制作人将发送另一个消息。。。
因为偏移量一个接一个地增加，而不是64的倍数
编辑-使用flink-kafka连接器
共有1个答案
消息由生产者批处理，以便最大限度地减少网络使用，而不是“作为批处理”写入Kafka的提交日志。您所看到的是Kafka正确完成的，因为每条消息都需要进行核算，即确定密钥/分区关系，附加到提交日志中，然后增加偏移量。除非完成前两个步骤，否则偏移量不会增加。
此外，还需要根据配置进行数据复制，并为收到的每条消息更新消息跟踪系统（以支持lag API）。
还要注意的是，批次。size参数考虑准备发送邮件的大小，该大小已预处理为1。压缩2。由您最喜欢的序列化程序序列化。
-
我有一个消费者作为生产者消费者模式的一部分： 简化： 如果我移除 通过将线程设置为睡眠，CPU使用率攀升到极高的水平（13%），而不是0%。 此外，如果我实例化该类的多个实例，则每个实例的CPU使用率都会以13%的增量攀升。 大约每分钟（可能每30秒）都会向BlockingCollection添加一个新的LogItem，并将适用的消息写入文件。 有没有可能线程以某种方式阻止了其他线程的运行，而系统
-
我试图写一个简单的例子，在CDI 2.0的JavaSE中使用生产者方法，但是遇到了一个问题。 以下是制作人： 这是注射点： 我通过如下方式引导运行示例： 我有一颗豆子。xml文件。 例外情况是： 线程“main”组织中出现异常。jboss。焊接例外情况。DeploymentException:WELD-001408:在注入点[BackedAnnotatedField]@InjectPrivate
-
我试图在我的Android应用程序中使用rx Java中的背压创建无限滚动。我希望它只调用请求的外部服务次数（在调用之后）。但在使用flatmap后，每个
-
本文向大家介绍在生产者中，何时发生QueueFullException？相关面试题，主要包含被问及在生产者中，何时发生QueueFullException？时的应答技巧和注意事项，需要的朋友参考一下 答：每当Kafka生产者试图以代理的身份在当时无法处理的速度发送消息时，通常都会发生QueueFullException。但是，为了协作处理增加的负载，用户需要添加足够的代理，因为生产者不会阻止。
-
我有一个生产者/消费者模式，如下所示 固定数量的生成器线程，每个线程写入它们自己的BlockingQueue，通过执行器调用 单个使用者线程，读取生产者线程 每个生产者都在运行一个数据库查询，并将结果写入其队列。消费者轮询所有生产者队列。目前，如果出现数据库错误，生产者线程就会死掉，然后消费者就会永远停留在产品队列中等待更多的结果。 我应该如何构造它来正确处理catch错误？
-
本教程演示了如何发送和接收来自Spring Kafka的消息。 首先创建一个能够发送消息给Kafka主题的Spring Kafka Producer。 接下来，我们创建一个Spring Kafka Consumer，它可以收听发送给Kafka主题的消息。使用适当的键/值序列化器和解串器来配置它们。 最后用一个简单的Spring Boot应用程序演示应用程序。 下载并安装Apache Kafka 要