kafka消费者如何反序列化不同的json对象？
Kafka消费者在处理不同的JSON对象时，需要进行反序列化操作。反序列化是将二进制数据或字符串转换为具体对象的过程。对于不同的JSON对象，可以采用以下几种方式进行反序列化：
- 手动反序列化：使用编程语言提供的JSON解析库，手动解析JSON字符串并将其转换为对应的对象。这种方式需要开发人员自行编写解析逻辑，较为灵活，但也需要处理错误和异常情况。
- 使用第三方库：许多编程语言都提供了成熟的JSON解析库，如Gson（Java）、Newtonsoft.Json（C#）、jsoncpp（C++）等。这些库可以根据JSON对象的结构自动进行反序列化，简化了开发过程。
- 使用Schema注册表：Kafka提供了Schema注册表，可以将JSON对象的结构注册到注册表中，并为每个JSON对象生成唯一的ID。消费者在反序列化时，可以根据ID从注册表中获取对应的Schema，然后使用Schema解析JSON数据。这种方式可以实现动态的反序列化，适用于处理多种不同结构的JSON对象。
在腾讯云中，可以使用腾讯云的消息队列 CMQ（Cloud Message Queue）来实现Kafka消费者的反序列化。CMQ提供了丰富的消息队列服务，支持多种消息格式，包括JSON。通过CMQ，可以将Kafka消息发送到CMQ队列中，然后使用CMQ提供的SDK进行反序列化操作。
腾讯云CMQ产品介绍链接：https://cloud.tencent.com/product/cmq
需要注意的是，以上答案仅供参考，具体的反序列化方式和工具选择应根据实际需求和开发语言来确定。
相关·内容
JSON消息正在由字符串使用者使用。我的produces发送两种类型的消息字符串和序列化的JSON
@KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.group-id}", containerFactory = "${kafka.string-listener-container-factory}")
public void consume(@NotNull ConsumerRecord<String, String> cr, @Payload String pa
浏览 0提问于2020-02-13得票数 1
我有一个非常简单的光束管道，它从Kafka主题读取记录，并将它们写入Pulsar主题： PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
p.apply(
KafkaIO.<Long, String>read()
.withTopic("<topic>")
.withBootstrapServers("<url>")
.withKeyDeseriali
浏览 12提问于2019-06-07得票数 1
回答已采纳
在阅读kafka文档时，我在kafka producer中发现了许多类型的序列化，但我对如何确定任何用例的序列化感到困惑。
浏览 18提问于2016-09-23得票数 0
我使用的是kafka-avro-serializer-6.0.0.jar。当我遇到反序列化事件的异常时，我的使用者会停止并且不会移动到下一个事件。这些问题通常是由生产者的错误引起的，并且由于使用新的avro模式注册服务器的问题而发生。
示例：
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 58
Caused by: java.lang.ClassCastException:
我可以修复异常，这不是问题所在。但为了修复消费者，我需要手动将每个偏移量重置为
浏览 0提问于2020-11-29得票数 0
我有一个spring boot web服务，它使用XML文件，向MQ生成JSON，但是，由于XSD schema中的标记和它们创建的后续XMLElementRef对象，我在解组XML时遇到了困难。作为一种变通方法，我制作了一个定制的JAXB element序列化器，ObjectMapper使用它将这些XML转换为JSON。
问题出现在Kafka的配置需要序列化程序。config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, null);不起作用，JsonSerializer也不起作用，而字符串序列化程序将所有的"转义为\"，
浏览 12提问于2018-12-12得票数 1
回答已采纳
我创建了一个Spring引导应用程序，它将消息发送到Kafka队列(使用spring集成kafka支持)。我想发送任意的json序列化对象。
有没有办法在我的spring boot应用程序中获得/注入json-de-/序列化程序？或者如何即席反/序列化一个对象？
应用序列化有哪些好的实践？
浏览 1提问于2018-01-29得票数 0
1回答
我正在学习Kafka，现在是Avro，为了保持发送方对象和接收方对象之间的一致性，我们保留了一个JSON (.avsc)。但我找不到任何简单的例子来说明如何使用它。有些例子是使用confluent (是Avro的汇合任务)，有些是通过Avro工具生成对象。到目前为止，我有一个卡夫卡的工作设置。
对象类
package com.example.kafka;
public class Hello {
String name;
String age;
public Hello(String name, String age) {
this.name = name;
this.
浏览 0提问于2019-08-18得票数 1
我将Spring Kafka侦听器初始化为
@Bean
public Map<String, Object> consumerConfig() {
final HashMap<String, Object> result = new HashMap<>();
result.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
result.put(GROUP_ID_CONFIG, groupId);
result.put(KEY_DESERIALIZER_CLASS_CONFIG,
浏览 0提问于2017-06-28得票数 9
如何使用kafka-console-producer在kafka topic中插入json文件数据？每个json数据集可以存储为一条消息吗？
示例：
{
"id": 1,
"first_name": "John",
"last_name": "Lindt",
"email": "jlindt@gmail.com",
"gender": "Male",
"ip_address": "1.2.3.4"
浏览 2提问于2019-02-13得票数 11
1回答
我正试着思考卡夫卡连接( Kafka )是如何工作的，但我无法理解一件特别的事情。
从我的阅读和观察来看，我理解Kafka Connect允许您使用Source连接器将数据发送到Kafka，使用Sink连接器从Kafka读取数据。最重要的是，Kafka从某种程度上抽象出了所有特定于平台的东西，而你所关心的就是有合适的连接器。例如，您可以使用PostgreSQL源连接器向Kafka写入，然后使用Elasticsearch和Neo4J Sink连接器并行读取卡夫卡的数据。
我的问题是:这种抽象是如何工作的？为什么不同人编写的源连接器和Sink连接器能够一起工作？为了从Kafka读取数据并将其写入任
浏览 2提问于2021-02-05得票数 0
当我尝试在kafka主题中使用字符串作为键时，我得到了以下错误。
18/05/14 17:08:26 ERROR async.DefaultEventHandler: Error serializing message for topic my_topic
java.lang.ClassCastException: java.lang.String cannot be cast to [B
at kafka.serializer.DefaultEncoder.toBytes(Encoder.scala:34)
at kafka.producer.async.DefaultEve
浏览 0提问于2018-05-15得票数 4
我有一个应用程序，它从Kafka中读取两个主题，第一个主题只是一个普通字符串，第二个主题是应该转换为DTO的JSON。 如果我使用RecordMessageConverter： @Bean
public RecordMessageConverter converter() {
return new StringJsonMessageConverter();
} 第一个主题(只是字符串)引发了一个将字符串转换为DTO的异常，因为StringJsonMessageConverter试图这样做。但第二个主题正常工作。 有什么建议，我该怎么解决呢？
浏览 31提问于2021-01-25得票数 0
回答已采纳
我正在用Apache + Kafka实现一个平台。我得到以下错误：
org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'responses': Error reading array of size 4416367, only 265 bytes available
at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:71)
这个错误的根源是什么？
解决办法是什么？
我在哪里可以找到更多的日
浏览 4提问于2016-09-14得票数 0
回答已采纳
2回答
我正在尝试使用spring云流绑定来实现一个自定义的Kafka分区工具。我只想自定义分区的用户主题，而不做任何与公司主题(卡夫卡将使用DefaultPartitioner在这种情况下)。
我的绑定配置：
spring:
cloud:
stream:
bindings:
comp-out:
destination: company
contentType: application/json
user-out:
destination: user
conten
浏览 0提问于2019-06-20得票数 3
我是阿帕奇卡夫卡的新手。我试图找出解决这个问题的办法，但失败了。生产者代码运行良好，数据存储在Kafka主题中。我正在使用JsonSerializer作为ValueSerializer在我的卡夫卡生产者配置。
NoramlizedEvent是一个简单的POJO，用于生产者和消费者。
我的制片人代码：
public void saveMessage(final IMMessage message) {
for (NormalizedEvent event :
message.getNormalizedEvents()) {
浏览 2提问于2021-05-10得票数 1
回答已采纳
1回答
我有两个弹簧引导应用程序。我试图映射到application.properties文件中，这样消费者就可以接收生产者发送的消息。我想补充一下，我使用的是一个CustomMessage：
public class CustomMessage {
private LocalDateTime timestamp;
private Integer sensor_id;
private Double measurement_value;
// getters and setters
}
我的生产者的application.properties文件：
# other propertie
浏览 2提问于2021-12-02得票数 2
回答已采纳
我目前有一个如下的开发环境。
弹簧-引导2.X
春季-kafka 2.5.5
Kafka版本2.5.1
生产者/使用者密钥序列化程序/反序列化器= String(De)serializer.class
生产者/消费者值序列化程序/反序列化程序= Json(De)serializer.class
当生产者发送Kafka消息时，定制的DTO(或POJO)通过JSON String转换为ObjectMapper.writeValueAsString(customDto)类型。
ProducerFactory<String，String>
KafkaTe
浏览 1提问于2021-04-08得票数 0
回答已采纳
我无法将数据作为逗号分隔的json对象发送给更大的数据集。
较小数据集的工作代码
df.selectExpr("CAST(collect_list(to_json(struct(*))) AS STRING) AS value") \
.write.format("kafka")\
.option("compression", "gzip")\
.option("kafka.bootstrap.servers", "localhost:9092"
浏览 4提问于2022-05-19得票数 0
对于运行kafka的相同机器集群，我有不同的主题来处理不同类型的消息，比如一个主题用于Protobuf，一个主题用于Avro，一个主题用于字符串消息。
这个设置会有什么问题吗？
浏览 0提问于2016-11-11得票数 1
3回答
只有为数不多的序列化程序可用，
org.apache.kafka.common.serialization.StringSerializer
我们如何创建自己的自定义序列化程序？
浏览 0提问于2016-10-20得票数 8
回答已采纳
Kafka Streams程序在什么情况下需要序列化/反序列化？假设我们有以下简单的程序：
KStream<k,v> stream = ...;
Kstream<k,v> stream2 = stream.filter( predicateA )
Kstream<k,v> stream3 = stream2.filter( predicateB)
stream3.to( topic );
具体地说，在filter的两次调用之间，"k“和"v”是被序列化/反序列化，还是单个数据点被作为本机对象传递？
浏览 0提问于2018-09-14得票数 0
我希望在同一个kafka消费者配置中有多个主题。
我尝试了以下选项，但不起作用。
使用
ContainerProperties containerProps = new ContainerProperties(gatewayPropertyConfigurer.getProperty("topic","topic1);
已为topic1创建单独的containerproperty。
如果你有什么建议，请告诉我。
浏览 0提问于2019-02-11得票数 0
我对kafka消费者有问题，它有时会抛出异常。
ERROR [*KafkaConsumerWorker] (Thread-125) [] Kafka Consumer thread 235604751 Exception while polling Kafka.: org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another membe
浏览 2提问于2020-06-10得票数 0
from kafka import KafkaConsumer
import json
import io
if __name__ == '__main__':
# consumer = KafkaConsumer(
# 'ldt_lm_mytable',
# bootstrap_servers = 'localhost:9092',
# auto_offset_reset = 'earliest',
# group_id = 'consumer_group_a&#
浏览 39提问于2021-01-27得票数 0
回答已采纳
在设置源连接器和接收器连接器之后，我会发现DATE类型Postgres列的问题。
错误："foo“列为日期类型，而表达式为整数类型
我检查了Avro模式，发现列foo被序列化为io.debezium.time.Date。
{
"default": null,
"name": "foo",
"type": [
"null",
{
"connect.name": "io.debezium.time.Date
浏览 1提问于2019-12-03得票数 2
回答已采纳
在我们的一个项目中，我们正在使用Kafka和AVRO在应用程序之间传输数据。将数据添加到AVRO对象，并对对象进行二进制编码以写入Kafka。我们使用二进制编码，因为与其他格式相比，它通常被认为是最小的表示形式。
数据通常是JSON字符串，当数据保存到文件中时，它最多使用10Mb的磁盘。但是，当文件被压缩(.zip)时，它只使用很少的KB。我们关心的是将这些数据存储在Kafka中，因此在写入Kafka主题之前尝试进行压缩。
当测量二进制编码消息的长度(即字节数组的长度)时，它与数据串的长度成正比。所以我假设二进制编码不会减少任何大小。
谁能告诉我二进制编码是否压缩数据？如果没有，我如何应用压缩
浏览 1提问于2014-11-03得票数 18
回答已采纳
我有一个场景，我有不同类型的消息从kafka producer中流式传输。
如果我不想针对不同的消息类型使用不同的主题，如何在spark-structured streaming消费者端处理它？
i.e. only one topic I want to use for different type of messages ...say Student record , Customer record....etc.
How to identify which message is been received from Kafka topic?
Please let me know how
浏览 10提问于2019-07-25得票数 1
2回答
我是Python的新手，从Kafka开始。我有一个需要发送和使用json消息的需求。为此，我使用kafka-python与卡夫卡进行交流。 #Producer.py
from kafka import KafkaProducer
import json
producer = KafkaProducer(bootstrap_servers='localhost:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8'))
producer.send('offering_new',
浏览 30提问于2019-04-05得票数 3
回答已采纳
我一直在尝试与kafka-avro-console-consumer从Confluent连接到我们的遗留Kafka集群，该集群是在没有Confluent Schema Registry的情况下部署的。我使用如下属性显式地提供了模式：
kafka-console-consumer --bootstrap-server kafka02.internal:9092 \
--topic test \
--from-beginning \
--property key.schema='{"type":"long"}' \
-
浏览 2提问于2018-09-19得票数 6
回答已采纳
我使用spring 2.3.2.RELEASE与spring -> 2.5.4 clients -> 2.5.0
我有以下简单的侦听器
@Slf4j
@Component
class SampleKafkaListener {
private final ObjectMapper objectMapper;
@Autowired
SampleKafkaListener(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
}
@KafkaListene
浏览 9提问于2020-08-25得票数 0
1回答
我们正在尝试使用Spring Kafka 2.6.x中的DLT功能。这是配置yml：
kafka:
bootstrap-servers: localhost:9092
auto-offset-reset: earliest
consumer:
key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
value-deserializer: org.springframework.kafka.support.serialize
浏览 1提问于2020-11-17得票数 0
2回答
我对Kafka很陌生，在尝试一个示例场景时，Kafka生产者以JSON格式向消费者发送用户详细信息。我曾访问过类似的问题，但我无法得到我所需要的答案。
我不面临任何问题，如果我运行任何生产者或消费者在一个终端，而另一个在春季启动。这个错误发生在无限循环中(当生产者和使用者都从不同的spring引导项目启动时)：
Consumer exception
java.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consi
浏览 9提问于2021-03-29得票数 1
回答已采纳
(在开始提问之前，我的英语可能不足以清楚地描述所有的事情。如果你不明白，请告诉我。)
我试图发送一个数据对象从一个弹簧项目(生产者)到B弹簧项目(消费者)卡夫卡。
问题是A和B中的数据对象有不同的类路径。因此，B项目数据类无法映射A项目的字段。
但是两个对象有相同的字段。所以我想从A项目中获取对象作为B项目的一个参数。
错误消息
Listener failed; nested exception is
org.springframework.kafka.support.serializer.DeserializationException: failed to deserialize; n
浏览 1提问于2021-08-30得票数 0
回答已采纳
11回答
在库更新之前的Spring / Kafka应用程序中，我使用以下类org.telegram.telegrambots.api.objects.Update将消息发送到Kafka主题。现在我使用以下org.telegram.telegrambots.meta.api.objects.Update。如你所见-他们有不同的包裹。
应用程序重新启动后，我遇到以下问题：
[org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1] o.s.kafka.listener.LoggingErrorHandler : Error whil
浏览 2提问于2018-08-04得票数 36
回答已采纳
1回答
我们有一些用c#和go语言开发的微服务。我们使用confluent-kafka进行消息传递，并面临一个问题: Go服务能够使用所有其他服务(无论是来自go还是c#/.net微服务)发布的消息。但是，.net微服务不是从go微服务而是从.net微服务获得发布的消息。 我们在windows平台上，使用confluent-kafka作为kafka服务器和.net客户端。在go中，我们使用的是kafka-go库(https://github.com/segmentio/kafka-go) 注意:在go中，kafka-go库允许以字节数组的形式发布消息。而在.net中，我们以纯文本形式发布消息。我不确
浏览 15提问于2019-04-19得票数 1
回答已采纳
对于Kafka主题，我可以使用confluent consumer api订阅和接收消息。对于Kafka Ktables，我可以使用REST API和http客户端进行订阅。所以我的问题是，也许可以不通过rest api而是confluent consumer api来订阅Kafka表？
以下是我订阅topic的方式：
using (var consumer = new ConsumerBuilder<Ignore, string>(conf).Build())
{
try
{
con
浏览 0提问于2019-07-31得票数 0
我查看了这些文档，发现如下：
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat
我的制片人的财产：
spring.kafka.producer.bootstrap-server=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-seriali
浏览 27提问于2022-09-16得票数 0
回答已采纳
1回答
我使用的是apache kafka版本2.6，并且只为avro序列化安装了confluent。我想知道提示我的kafka消费者只使用在服务器级别本身正确序列化的记录所需的配置更改。
浏览 18提问于2021-11-30得票数 0
1回答
我需要一个建议，什么是正确的卡夫卡结构使用我的项目和原因。
我的项目是为投资机器人管理创造一个平台。非常高水平-你可以编码几个投资策略，并将它们上传到平台上，它们将实时执行，提供分析和性能的实时信息。这些策略是从4个数据流中输入信息的。当他们阅读四个不同的卡夫卡主题时，这些数据被传递给策略。这个kafka主题直接从交换websocket接收信息。在任何给定的时间，平台中都有一个动态的机器人数量。
我所做的是:使用图像卡夫卡-wurmeister和动物园，初始化卡夫卡初始化所有的卡卡主题，我将需要事先。我将所需的数据推送给Kafka，方法是将所有信息生成到以下主题：
payloads = [
浏览 3提问于2020-05-10得票数 0
回答已采纳
我需要在发布/订阅模式下调用Kafka consumer 1000次。据我所知，为了让kafka在发布/订阅模式下工作，我需要给每个消费者一个新的groupId ( props.put("group.id"，kafka但是当我这样做时，如果两个消费者线程在同一毫秒内访问消费者，就会出现问题。这个问题该如何解决呢？
浏览 1提问于2018-05-13得票数 1
我使用的是spring-cloud-stream-kafka，我不明白如何在使用事务时更改key.serializer属性。这是我的配置： spring:
cloud:
stream:
bindings:
accountSource:
contentType: application/*+avro
destination: account
kafka:
binder:
brokers: ${KAFKA_BOOTSTRAP_ADDRESSES}
t
浏览 13提问于2019-06-18得票数 0
回答已采纳
我对Apache Kafka非常陌生。我在C#.NET /Windows platform上工作。我能够使用Kafka的配置设置，并使用命令提示符发送和接收消息。
我使用运行了c#程序，并且能够生成和接收消息。Qn1:我正在尝试读取一个json文件，根据json文件中的行数，有没有办法从C#中动态改变json值的值？
谢谢先生
浏览 1提问于2019-01-16得票数 0
1回答
下面是如何使用Kafka作为发布订阅框架的一个很好的例子：
我正在拼命寻找一个类似的例子，展示如何使用kafka作为队列框架。这意味着一旦一个客户端检索到消息，任何其他客户端都不应该访问它。
浏览 0提问于2014-11-11得票数 0
1回答
我正在将一个java对象序列化为JSON，如下所示 ObjectMapper mapper = new ObjectMapper();
String json = mapper.writeValueAsString( serializableObject ); 当我在控制台中打印它时，我可以正确地看到它。 {
"topicName" : "LVS",
"sellerId" : "TS",
"orderId" : "123456",
"lineIds" : [ {
"
浏览 82提问于2020-05-02得票数 0
回答已采纳
我需要使用带有org.springframework.kafka依赖关系的Apache使用Spring应用程序中的json对象消息，但我得到了以下错误：
***************************
APPLICATION FAILED TO START
***************************
Description:
Parameter 1 of method kafkaListenerContainerFactory in org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenCon
浏览 2提问于2020-03-22得票数 3
回答已采纳
我已经为Kafka的生产者端找到了很好的参考架构细节，但是关于端到端模型的细节很少，比如如何从主题中实际消费消息，如何管理订阅，订阅在哪里持久化(我假设作为组协调器上的主题？)，以及反序列化发生的时间。
有没有人有任何细节来帮助确定代理和消费者之间的详细架构？
浏览 2提问于2019-07-04得票数 0
这就是场景:我知道，使用与Spring相关的最新API (比如kafka 2.10)，我们可以这样做：
@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic", partitions = { "0" }) })
@KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic",
浏览 0提问于2017-04-17得票数 1
我在Kafka Streams应用程序中使用了以下JSON配置，
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.String().getClass());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
GenericJSONSerde.class);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
浏览 3提问于2019-10-25得票数 3
回答已采纳
2回答
我在一个Kafka使用者上有以下spark /Streaming，当批大小达到特定大小N时，如何指定获取应该是有条件的，否则使用者应该在处理之前缓冲元素，所以每当我想要执行我的逻辑时，都可以保证我有一个大小为N的Dataset<VideoEventData>：
Dataset<VideoEventData> ds = spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", prop.getProperty("
浏览 3提问于2017-11-27得票数 1
回答已采纳
1回答
我需要Serialize或Deserialize ，在我的项目中，任何类型的Java对象都可能是Integer/ String、<T>或User或Account。可能有一个以上的类型的，我不确定在配置Kafka、Producer和Consumer时使用哪一个。
有JsonSerializer、JsonDeserializer和StringSerializer/DESer以及更多的类型。
我读过
那么，我如何得出一个结论，用哪一个呢？谢谢。
浏览 0提问于2021-05-06得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
