如何使用Kafka控制台生产者处理压缩数据
Kafka是一种分布式流处理平台，用于高吞吐量、可持久化、可扩展的发布和订阅消息系统。Kafka控制台生产者是Kafka提供的一个命令行工具，用于向Kafka集群发送消息。
要使用Kafka控制台生产者处理压缩数据，可以按照以下步骤进行操作：
- 安装Kafka：首先需要安装Kafka集群，并确保集群正常运行。可以参考腾讯云的Kafka产品文档（https://cloud.tencent.com/document/product/597）了解如何在腾讯云上部署Kafka集群。
- 创建主题：在Kafka集群上创建一个主题，用于存储发送的消息。可以使用Kafka提供的命令行工具或者腾讯云的Kafka管理控制台（https://console.cloud.tencent.com/ckafka）进行主题创建。
- 启动Kafka控制台生产者：在命令行中执行以下命令启动Kafka控制台生产者：
- 启动Kafka控制台生产者：在命令行中执行以下命令启动Kafka控制台生产者：
- 其中，
<broker-list>是Kafka集群的地址列表，<topic-name>是要发送消息的主题名称，<compression-codec>是指定要使用的压缩算法，例如gzip、snappy等。 - 发送压缩数据：在Kafka控制台生产者启动后，可以在命令行中输入要发送的消息。如果要发送压缩数据，可以在输入的消息中使用压缩算法支持的格式，例如使用gzip压缩的消息可以是二进制格式。
使用Kafka控制台生产者处理压缩数据的优势包括：
- 减少网络传输量：压缩数据可以减少消息在网络中的传输量，提高传输效率。
- 节省存储空间：压缩数据可以减少消息在磁盘上的存储空间占用。
- 提高传输速度：压缩数据可以减少消息在网络中的传输时间，提高传输速度。
Kafka控制台生产者处理压缩数据的应用场景包括：
- 大规模数据传输：当需要传输大量数据时，使用压缩数据可以减少网络传输量，提高传输效率。
- 节省存储空间：当需要存储大量数据时，使用压缩数据可以减少存储空间占用，降低存储成本。
腾讯云提供的相关产品和产品介绍链接地址如下：
- 腾讯云CKafka：https://cloud.tencent.com/product/ckafka
- 腾讯云消息队列CMQ：https://cloud.tencent.com/product/cmq
相关·内容
2回答
我正在使用kafka节点()使用者来检索数据。我想我得到的数据被快速压缩了。我如何在得到数据后解压缩数据。我尝试使用node ()来解压缩数据，但是它没有工作。
库中是否有将压缩设置为none的选项？
有人使用kafka节点库从kafka获取数据？
谢谢你，钱杜
浏览 3提问于2014-06-27得票数 3
回答已采纳
我在Kafka中配置了3个代理，运行在不同的端口上，.I正在使用spring cloud stream kafka
brokers: localhost:9092,localhost:9093,localhost:9094.
我正在创建一个数据管道，以获得连续的数据流.I正在与运行.Till的3个代理在kafka主题中存储数据流现在没有问题.My问题假设假设3个代理停机了5分钟，那么到那时我无法获取有关kafka的数据主题.There将丢失数据5分钟.From spring启动我将收到警告
2020-10-06 11:44:20.840 WARN 2906 --- [ad | produc
浏览 0提问于2020-10-06得票数 1
2回答
我对Kafka2.6.0中的消息大小配置有点困惑。但让我们来说说故事：
我们正在使用一个由3个节点组成的Kafka集群。到目前为止，对于消息的标准配置。"zstd压缩“被激活。
相关的代理配置很简单：
compression.type=zstd
在这一点上，生产者配置也很简单：
compression.type=zstd
现在，我们希望将具有8M字节的消息放到特定的主题中。此数据的压缩大小仅为200千字节。
如果我将这些数据放入主题中，就会发生以下错误：
sudo /opt/kafka/bin/kafka-console-producer.sh --topic XXX --produc
浏览 5提问于2020-11-05得票数 5
回答已采纳
我用CONNECT_CONNECTOR_CLIENT_CONFIG_OVERRIDE_POLICY=All在本地机器上运行了Kafka，用"producer.override.compression.type": "snappy"运行了连接器。我的理解是，它将使Debezium连接器对它生成的JSON消息使用Snappy压缩。当我运行Kafka控制台使用者时，我能够看到JSON的有效负载。
有人能帮我理解减压在哪里发生吗?为什么我在控制台里看不到胡言乱语？
浏览 3提问于2022-02-23得票数 0
回答已采纳
1回答
Spring循环发送消息
、、、
我使用spring依赖版本2.8.1，配置如下：
kafka:
bootstrap-servers: localhost:9092
producer:
acks: all
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
buffer-memory: 16384
我试图向一个主题发送大约65.
浏览 7提问于2022-02-08得票数 1
2回答
无法向Kafka发送大消息
、、、、
我想从producer向Kafka发送一条大消息，所以我更改了下面的属性。
代理(server.properties)
replica.fetch.max.bytes=317344026
message.max.bytes=317344026
max.message.bytes=317344026
max.request.size=317344026
生产者(producer.properties)
max.request.size=3173440261
消费者(consumer.properties)
max.partition.fetch.bytes=327344026
fetch.mes
浏览 0提问于2018-08-09得票数 3
Kafka按照buffer.memory将数据保存在缓冲区中(在我的例子中为32 MB )。卡夫卡是否在达到32 MB的限制时将记录写入主题，或者是否也有相关的时间？
浏览 2提问于2020-11-27得票数 3
我试图从Kafka (用Java语言)中读到一个主题，但这个例外总是在启动：
kafka.common.UnknownCodecException: 3 is an unknown compression codec
at kafka.message.CompressionCodec$.getCompressionCodec(CompressionCodec.scala:26)
at kafka.message.Message.compressionCodec(Message.scala:213)
at kafka.message.ByteBufferMessageS
浏览 5提问于2016-03-08得票数 2
回答已采纳
在kafka中启用压缩的下列方法有什么区别：
方法1:使用以下命令创建一个主题：
bin/kafka-topics.sh --create --zookeeper localhost:2181 --config compression.type=gzip --topic test
方法2:在Kafka生产者客户端API中设置属性compression.type = gzip。
当使用方法1时，我得到了更好的压缩和更高的吞吐量。
如果我使用方法1，是否意味着压缩发生在代理端，而在方法2中，消息在生产者端被压缩，然后发送给代理？
浏览 0提问于2018-02-07得票数 29
下面是我的生产者配置，如果您看到他们的压缩类型是gzip，即使我提到了为什么消息没有发布并且失败的原因是压缩类型
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, edi856KafkaConfig.getBootstrapServersConfig());
props.put(ProducerConfig.RETRIES_CONFIG, edi856KafkaConfig.getRetriesConfig());
props.put(ProducerConfig.BATCH_SIZE_CONFIG, edi8
浏览 0提问于2017-12-07得票数 1
对卡夫卡来说是个新手。
今天我运行了kafka-console-producer.sh的一些命令，我使用了键值对消息
bin/kafka-控制台-Producer.sh-主题生产者-演示-属性"parse.key=true“-属性"key.separator=:”-broker-list kafka1:9092
我只是好奇我们在哪里能像我搜索的那样找到生产者的财产
**"parse.key=true"**和
**"key.separator=:"** in
没有像这样的参数，这是这个自定义参数或预定义的，因为我必须更多地使用这个参数，并
浏览 3提问于2021-06-14得票数 0
当使用时，我可以通过设置我的kafka生产者的kafka.compression.codec属性来设置编解码器。
假设我在我的生产者中使用了snappy压缩，当我使用某个kafka消费者消费kafka的消息时，我应该做一些事情来解码来自snappy的数据还是它是kafka消费者的内置功能？
在中，我找不到任何与kafka consumer中的编码相关的属性(它只与生产者相关)。
有人能把这个清理一下吗？
浏览 0提问于2013-11-10得票数 9
回答已采纳
如果我们没有提到kafka producer上的压缩类型，如果我们在broker端提到它...性能是如何受到影响的，主题端压缩在哪些批量大小上起作用？
浏览 13提问于2018-08-09得票数 0
我们有一个前层，它只接收消息并写入Kafka主题以进行后端处理。我们以非常高的速率发送消息；每天我们处理10亿条消息。我们有一个接受消息并写入Kafka生产者实例的线程池。在这里，我只创建了一个生产者(单个实例)，它在多个线程之间共享。
最近，我观察到90%的线程处于阻塞状态。我发现Kafka是按顺序发送数据的。在Kafka Java驱动程序的producer.send()方法中有一个synchronized块：
def send(messages: KeyedMessage[K,V]*) {
**lock synchronized {**
if (hasShutdown.
浏览 0提问于2018-08-08得票数 0
1回答
我试图使用Kafka连接来使用独立模式编写数据。我正在编写数据的主题是有多个分区。但是，数据只被写入其中一个分区。当我启动多个使用者控制台时，数据只会打印到其中一个。另一个使用者控制台只有在第一个用户控制台关闭后才能获得任何数据。我无法确定需要在配置文件中进行哪些更改才能使其写入多个分区。
这是standalone.properties
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connec
浏览 1提问于2017-06-26得票数 0
回答已采纳
是否有人使用Apache Kafka在每个服务都有自己的数据库的微服务体系结构中维护数据完整性？我一直在搜索，有一些帖子提到了使用Kafka，但我正在寻找更多的细节，比如Kafka是如何使用的。您是否必须为生产者和消费者编写代码(例如，对于作为生产者的客户数据库和作为消费者的订单数据库，以便如果在客户数据库中删除了一个客户，那么订单数据库需要以某种方式知道这一点，这样它也会删除该客户的所有订单)。
浏览 1提问于2019-12-06得票数 1
我想用kafka压缩消息。因此，我在生产者端启用了压缩类型为gzip，并且没有在server.propeties上做任何更改。 config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip"); 这是我在生产者配置中添加的代码，但当我向主题发送消息时，当我在消费者端读取消息时，消息并未压缩。有没有人能建议一下在kafka中如何在生产者端做消息压缩？我使用的是JAVA和Springboot。 提前谢谢。
浏览 85提问于2020-11-23得票数 0
我使用的是spring boot 2.1.9和spring Kafka 2.2.9。
我的spring boot应用程序在过去的两周里一直处于闲置状态，它还在运行，但是仍然没有人访问API。但是当我今天开始使用的时候，对于第一个请求，它给了我下面的错误。
2019-12-18 14:43:03.908 ERROR [xxxx-account-service,681c8de32a0d1127,681c8de32a0d1127,false] 10 --- [nio-8080-exec-5] o.s.t.i.TransactionInterceptor : Application
浏览 0提问于2019-12-19得票数 2
1回答
如何处理卡夫卡中的大消息，比如超过20 to的消息等等。
2009-03- 13 08:59:10,923错误当发送消息到主题测试时，键:13字节，值: 11947696字节与错误：(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.RecordTooLargeException:请求包含一条大于服务器将接受的最大消息大小的消息。
2019-03-1303:59:14,478在用键向主题测试发送消息时出错: 13字节，值: 11947696字节带有错误
浏览 0提问于2019-03-13得票数 1
回答已采纳
在使用者行中获取消息中的错误
代码：
使用者=KafkaConsumer(主题，group_id=groupid，bootstrap_servers=host)用于使用者中的消息：
错误:关于使用者中的消息:文件"\Python\Python38-32\lib\site-packages\kafka\consumer\group.py"，第1192行，next返回self.next_v2() self.next_v2行，next_v2返回下一步(self._iterator)文件"\Python\Python38-32\lib\site-packages\kafka\c
浏览 4提问于2020-05-15得票数 0
4回答
在我读过的所有Kafka教程中，他们都提到"Kafka分区是不可变的“。但是，我也从这个站点上读到，Kafka会不时地删除分区中的旧消息(取决于您在log- that命令中设置的保留时间)。您可以从下面的屏幕截图中看到，在删除分区中的重复键之后，分区中的数据明显发生了变化：
所以我的问题是，说"Kafka分区是不变的“到底是什么意思？
浏览 8提问于2019-10-03得票数 1
回答已采纳
2回答
我正在尝试理解一个卡夫卡制作人是如何工作的。下面是我用来发送消息的python生成器代码。我首先启动了Kafka控制台使用者，然后运行python代码。 from confluent_kafka import Producer
from Product import Product
from faker import Faker
if __name__ == '__main__':
config = {
"bootstrap.servers":"localhost:9092"
}
produce
浏览 37提问于2020-11-20得票数 0
2回答
实时事件处理
、、
我真的很想为我下面的场景得到一个架构解决方案。
我有一个事件源(比如油井中的传感器，大约50000个)，它向服务器产生事件。在服务器端，我想以这样的方式处理所有这些事件，即来自传感器的关于最新湿度、温度、压力...etc的信息将被存储/更新到数据库中。
我把flume和kafka搞混了。
有没有人能用建筑术语来解释一下我的简单场景。
我不想将事件存储在某个地方，因为我已经用最新的值更新了数据库。
我真的需要spark，(flume/kafka) + spark，来满足处理端吗？
我们可以在没有水槽的情况下使用flume进行任何类型的处理吗？
浏览 13提问于2017-06-29得票数 1
1回答
卡夫卡生产商正在发送.gz文件，但无法解压缩和读取消费者端的文件。获取错误为"IOError:而不是gzipped文件“
生产者-bin/kafka-控制台-Producer.sh-broker-list localhost:9092 -主题机场<~/下载/Stocks.json.gz
消费者-
import sys
import gzip
import StringIO
from kafka import KafkaConsumer
consumer = KafkaConsumer(KAFKA_TOPIC, bootstrap_servers=KAFKA_BROKERS)
浏览 0提问于2018-05-08得票数 2
1回答
我正在使用Alpakka kafka和scala应用程序。我的Kafka在docker内部运行，我正在尝试使用我的代码在Kafka producer上发布消息。我的代码如下
def sendMsg(xmlFile: String): Future[Done] = {
futureToFutureTry {
val producer = SendProducer(producerSettings)
producer.send(new ProducerRecord("topic_name", "Key", xmlFile)).map
浏览 4提问于2020-07-17得票数 1
1回答
如果Kafka生产者使用snappy压缩，那么做代理度量，例如，
BrokerTopicMetrics.topic.{topic}.BytesInPerSec
BrokerTopicMetrics.topic.{topic}.BytesOutPerSec
表示压缩或未压缩字节？
浏览 5提问于2022-02-15得票数 1
我有一个需要向Kafka控制台生产者发送键值消息的用例。那么如何通过Kafka Console Producer命令来实现这一点呢？
浏览 1提问于2020-05-29得票数 20
回答已采纳
2回答
我想使用Python Kafka连接器向Kafka发送数据。当我从pyspark shell运行代码时，一切工作正常。但是，当我以spark-submit身份运行它时，消息不会被发送。日志中没有错误，程序执行显示为成功。但这些消息不会发送给卡夫卡。
import json
import datettime
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='XXX.XX.XXX.XXX:9092')
end = datetime.datetime.now().isoformat
浏览 0提问于2018-08-17得票数 0
我有一个包含3个kafka节点和3个zk节点的kakfa集群。
生产者在AWS机器上，试图在我的内部网服务器上运行的kafka集群上推送数据。
当通过控制台创建主题(JOB_AWS_14)时
Partition: 0 Leader: 1 Replicas: 1 Isr: 1
它工作得很好。
但是在创建主题(JOB_AWS_8)时使用
Partition: 0 Leader: 2 Replicas: 2 Isr: 2
它不起作用。
哪些设置出错，以及如何纠正它。
请帮帮我。
# bin/kafka-topics.sh --describe --
浏览 0提问于2016-02-11得票数 0
2回答
在我的应用程序中，我们使用Kafka高级用户，如果生产者和使用者使用java对数据进行压缩和解压缩，它将使用解压缩数据，而不会出现任何问题。
如果生产者使用librdkafka C++ API进行压缩(无论是snappy还是GZIP)，会发生什么？java使用者是否像上面提到的那样透明地解压缩。消费者端的获取大小会发生什么？这也是透明处理的吗。
如果卡夫卡消费者是使用简单的消费者模型设计的，会发生什么？我们是否必须显式地解压缩来自生产者的压缩数据(假设这里使用的是librdkafka C++ API )。
我在想，在发生压缩的情况下，高级别的使用者可能无法在生产者端使用librdkafka C
浏览 0提问于2015-01-22得票数 1
回答已采纳
1回答
Kafka和IIDR CDC
、、、、
我正在尝试使用：DB2--IBM -Kafka来构建CDC管道，并试图找到正确的方法来设置这个管道。我试过在下面-
1.在prem上的linux上安装一个3节点kafka集群
2.使用- setup-iidr-11.4.0.1-5085-linux-x86.bin文件在linux上安装IIDR软件。CDC实例已经启动并运行。
各种在线文档建议安装“IIDR管理控制台”来配置源数据存储和CDC服务器配置，并配置Kafka订阅配置来构建管道。
目前，我没有安装管理控制台。很少有人问这个问题-
1.在设置kafka-CDC管道时，是否有任何替代IBM管理控制台的方法？
2.如何获得IIDR管理控制台
浏览 2提问于2018-11-18得票数 1
回答已采纳
1回答
我在世界各地有许多应用程序可以将数据发送到我的服务器。所有的数据都是使用Kafka流产生的，作为avro进入Kafka主题，有40个分区，我想防止重复数据。在我的情况下，应用程序可以多次发送相同的消息(甚至在几分钟后)，我想知道Kafka是否可以帮助我阻止它。有没有“散列”消息和防止产生两次卡夫卡的选项？
我只读过一次，等等。这似乎支持了制片人未能生产的情况。
我希望Kafka会在有限的时间内(比如说10分钟)防止重复的消息。
浏览 0提问于2019-06-19得票数 0
我正在使用@SendTo("example")注释向主题发送消息，但我不知道如何测试它。我打开一个控制台生产者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic example
在控制台中似乎可以工作，但是我在生产者中没有收到任何消息
gid: trace= span= [kafka-producer-network-thread | producer-1] org.apache.kafka.clients.Metadata Cluster ID: uX-1vV4bSFup6t_Sh_LNCg
浏览 9提问于2020-04-09得票数 1
回答已采纳
1回答
我们使用Kafka流将图像文件从一个web服务发送到另一个web服务，以保持持久化。在此过程中，我们将图像转换为字节数组，然后将它们转换为Base64编码的字符串，并将其与其他文件元数据一起作为消息发送。由于大的图像文件大小，我们的卡夫卡正在下降。
有什么更好的选择来流这些图像通过卡夫卡和文件元数据？
浏览 0提问于2017-01-11得票数 0
回答已采纳
1回答
我正在使用KSQL处理卡夫卡数据。我可以创建持久的kafka流，但是这个流的输出主题不是压缩的。如何在创建立体声时定义压缩类型。
CREATE STREAM TARGET_TOPIC WITH **(COMPRESSION='SNAPPY')** AS SELECT * FROM SOURCE PARTITION BY ID
我期待提供像上面那样的压缩配置，但是上面的语法是不正确的。如果在ksql中压缩是可能的，请在这方面提供帮助。
谢谢
浏览 2提问于2019-09-25得票数 0
我想实现对kafka消息的压缩。为此，必须设置compression.type。应该在哪个属性文件中设置它？
浏览 0提问于2018-05-04得票数 1
我们使用Kafka来存储由集群中的节点生成的消息，并分发到集群中的所有节点，我让它主要使用akka-streams，但有几个问题我必须解决。这是有一些限制的。
首先，消息必须由集群中的每个节点使用，但只由一个节点生成。我知道我可以为每个节点分配一个组id，这可能是它的节点ID，这意味着每个节点都会收到消息。已经排序好了。但这里有一些问题。
数据是非常短暂和相当大的(仅低于一兆字节)，不能进一步压缩或分解。如果有关于这个主题的新消息，那么旧消息几乎是垃圾。如何才能将主题限制为当前最多只有一条消息？
考虑到数据是节点启动所必需的，我需要消费关于该主题的最新消息，无论我以前是否消费过它，并且希望不会
浏览 0提问于2017-12-06得票数 0
1回答
错误
Expiring 2 record(s) for catering-0:120012 ms has passed since batch creation
首先我使用了配置
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
ports:
- "2181:2181"
restart: always
kafka:
image: wurstmeister/kafka
container_name: kafka
浏览 9提问于2022-07-11得票数 0
在Kafka中，您可以在生产者上设置属性来压缩键和值。
compression.codec此参数允许您为此生成器生成的所有数据指定压缩编解码器。有效值是"none“、"gzip”和"snappy“。
当使用汇流平台时，我可以使用KafkaAvroSerialization序列化程序。如果我定义avro.codec，那么使用卡夫卡的压缩或相反的压缩有什么意义吗？
avro.codec用于压缩块的压缩编解码器的名称，作为字符串。实现需要支持下列编解码器："null“和"deflate”。如果没有编解码器，则假定为"null“。下面将更详细
浏览 2提问于2015-09-07得票数 2
回答已采纳
2回答
我怎么能从卡夫卡那里读得更快
、、、、
我创建了一个新的Kafka服务器(我创建了一个带有一个分区的代理)，我成功地使用了java代码在这个服务器上生成和使用，但我并不满足于我作为消费者每秒读取的事件数量。
我已经使用了下列消费者设置：
AUTO_OFFSET_RESET_CONFIG = "earliest"
FETCH_MAX_BYTES_CONFIG = 52428800
MAX_PARTITION_FETCH_BYTES_CONFIG = 1048576
MAX_POLL_RECORDS_CONFIG = 10000
pollDuration = 3000
但是，无论我向每个设置输入了什么值，结果都是相同的。
浏览 0提问于2019-08-08得票数 2
回答已采纳
KafkaStream在很长一段时间后无法生成数据。(超过设定的过期时间)
在记录错误消息后，甚至KafkaStream也死了。
例外情况如下：
org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:784)
at o
浏览 0提问于2018-12-30得票数 2
1回答
我试图复制中给出的步骤。在尝试时，给出了一个Kafka Consumer和Kafka Producer python代码，我可以在python交互式终端上运行这些代码，并且消费者控制台能够给出输出，但是如果我将它们传递到python (*.py)中，它就不会消耗任何东西。
消费者
from kafka import KafkaConsumer
consumer = KafkaConsumer('sample')
for message in consumer:
print (message)
生产者
from kafka import KafkaProducer
pro
浏览 1提问于2019-05-06得票数 1
回答已采纳
1回答
我们的团队正在将Kafka与一个用于实时显示数据的烧瓶应用程序集成在一起，但我们也想显示卡夫卡的历史数据。
这样，我们就可以使用来自特定主题的所有消息，并将数据显示给用户。然而，当我们设置Avro使用者来轮询整个主题的消息时，我们每分钟只能消耗100 k-200 k消息，这太慢了，因为我们每个主题大约有250万条消息。即使当我们用相同的组id设置upp多个使用者时，我们仍然没有多少性能改进。
关于如何更快地获取卡夫卡主题的所有信息，有什么建议吗？还是最好将数据保存到数据库中，然后从数据库中查询数据？
我们的消费者：
c = Consumer({
'bootstrap.serve
浏览 5提问于2020-11-15得票数 1
回答已采纳
我用java构建了一个kafka应用程序，它具有：
是一家生产卡夫卡唱片的生产商--卡夫卡是一种卡夫卡流，使用这些记录，对其值应用一些(时间窗口和状态存储)操作，并将它们发送回kafka ，消费者消费这些转换后的值，并将它们写入数据库。
我是测量卡夫卡时间戳的生产者(它被卡夫卡流消费)和消费者记录(它被消费者消费)之间的时间差。所以，基本上，当制作人的唱片被创建，当这张唱片被流转换，并被送回卡夫卡时。最后，我取了数据库中每个时差的平均值。
由于任何原因，当我向主题添加更多流节点和更多分区时，这种时间差就会增大。实际上，我预计时差会减小。现在，我想知道我是否做错了什么，或者通过增加节点数量，数据
浏览 3提问于2020-12-03得票数 2
1回答
我们安装了8个节点的kafka集群和kafka管理器。
我们正在通过新的遗迹进行监测。
new Relic和kafka管理器都报告kafka拒绝字节。我找不到原因。
在broker日志中没有错误行。
JMS BEAN - JMX/kafka.server/BrokerTopicMetrics/BytesRejectedPerSec/OneMinuteRate
Kafka配置-
auto.create.topics.enable=false
auto.leader.rebalance.enable=true
broker.id=180
controlled.shutdown.enab
浏览 0提问于2020-01-10得票数 2
我需要更改Kafka设置中的值序列化程序/反序列化程序(出于测试目的，我一直在使用IntegerSerializer/IntegerSerializer)。使用JAVA API，它可以完全按照预期工作；但是，在使用控制台工具时，它似乎不能正常工作。
我所做的所有故障排除都让我得出了一个结论:kafka控制台生产者似乎忽略了序列化选项。我尝试了--property value.serializer、--producer-property value.serializer和--value-serializer，并在配置中使用--producer.config参数对其进行设置。
它不仅没有将数据序列
浏览 20提问于2017-06-28得票数 5
回答已采纳
我试图在nodejs中使用Kafka节点创建kafka制作人的消息块，但没有得到任何解决方案。因此，现在我正在使用Java创建一个Kafka流生成器，我需要发送大小大于1MB的大型消息。我怎样才能在卡夫卡制作者中创造出大量的信息，并消耗同样的信息呢？
浏览 0提问于2019-09-12得票数 0
4回答
同步来自多个数据源的数据
、、、、
我们的团队正在尝试构建一个预测维护系统，其任务是查看一组事件，并预测这些事件是否描述了一组已知的异常。
我们正处于设计阶段，目前的系统设计如下：
这些事件可能发生在IoT系统的多个源(如云平台、边缘设备或任何中间平台)上。
事件由数据源推入消息队列系统(目前我们选择了Apache )。
每个数据源都有自己的队列(Kafka主题)。
从队列中，数据由多个推理机(实际上是神经网络)消耗。
根据特征集的不同，推理引擎将订阅多个Kafka主题和来自这些主题的流数据，以持续输出推理。
总体架构遵循单一责任原则，这意味着每个组件将彼此分离，并在单独的Docker容器中运行。
浏览 0提问于2019-05-27得票数 8
回答已采纳
3回答
我是卡夫卡的新手。我有关于kafka配置的问题。
我想使用如下所示的独立服务器，
server1:卡夫卡生产者server2:卡夫卡经纪人，卡夫卡消费者，动物园管理员
但是，我不能向broker发送消息。我收到了这个错误消息。
在控制台生产者(Server1)上，控制台标准输出错误消息`
[2016-05-24 16:41:11,823] ERROR Error when sending message to topic twitter with key: null, value: 3 bytes with error: Failed to update metadata after 6000
浏览 1提问于2016-05-25得票数 6
我正在开发一个Spring引导应用程序。该应用程序充当Kafka-producer，其中生产/发布消息并不是很重要。我正在使用spring-kafka依赖项，其配置如下 @Bean
public ProducerFactory<String, Object> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfi
浏览 245提问于2021-08-12得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
