如何在 Spring Cloud Stream App 中反序列化 MessagePack答案

【问题标题】：How to deserialize MessagePack in Spring Cloud Stream App如何在 Spring Cloud Stream App 中反序列化 MessagePack
【发布时间】：2020-06-01 20:58:16
【问题描述】：

我正在尝试使用 Spring Cloud Stream 创建一个 Kafka 流应用程序，但我正在努力反序列化输入消息，其值已使用 MessagePack 编码。

这是我目前所得到的：

// TransactionApplication.java

@SpringBootApplication
public class TransactionApplication {

  public static void main(String[] args) {
    SpringApplication.run(TransactionApplication.class, args);
  }

  public static class TransactionConsumer {

    @Bean
    public Serde<Transaction> transactionSerde() {
      ObjectMapper mapper = new ObjectMapper(new MessagePackFactory());
      return new JsonSerde<Transaction>(mapper);
    }

    @Bean
    public Consumer<KStream<String, Transaction>> process() {
      return input -> input.foreach((key, value) -> {
        System.out.println("Key: " + key + " Value: " + value);
      });
    }
  }
}

// Transaction.java

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Transaction {
  String item;
  Number amount;
}

我收到了错误：

java.lang.IllegalStateException：标头中没有类型信息，也没有提供默认类型。

我的 application.yml 是：

spring.cloud.stream:
  bindings:
    process-in-0:
      destination: transactions
  kafka:
    streams:
      binder:
        applicationId: transactions-application
        configuration:
          commit.interval.ms: 100

在我的 applicaton.yml 的 configuration 节点下包含 spring.json.value.default.type: com.example.Transaction 后，我又遇到了另一个错误。见下文。

Caused by: org.apache.kafka.common.errors.SerializationException: Can't deserialize data [[123, 34, 105, 116, 101, 109, 34, 58, 32, 34, 112, 114, 105, 118, 97, 116, 101, 32, 106, 101, 116, 34, 44, 32, 34, 97, 109, 111, 117, 110, 116, 34, 58, 32, 53, 48, 50, 125]] from topic [transactions]
Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance of `com.jackdry.processors.json.Transaction` (although at least one Creator exists): no int/Int-argument constructor/factory method to deserialize from Number value (123)
 at [Source: (byte[])"{"item": "private jet", "amount": 502}"; line: -1, column: 0]

【问题讨论】：

applicaton.yml 是这里的拼写错误，还是在您的系统中也有错误的名称？

标签： java apache-kafka apache-kafka-streams spring-kafka spring-cloud-stream

【解决方案1】：

您需要向反序列化器提供一个提示，告诉它从编码的有效负载中创建什么对象。

如果记录是由 Spring JsonSerializer 创建的，则提示位于标头中。

如果没有，您必须在流配置中提供提示。

您需要出示您的application.yml/properties。

【讨论】：

在活页夹configuration:节点下，添加spring.json.value.default.type=com.foo.Transaction。
no int/Int-argument constructor我对MessagePack不熟悉；通常，Jackson 需要 no-arg 构造函数 Foo() 来创建 Foo。但在这种情况下，它似乎正在寻找一个构造函数（Transaction(int someInteger)）并且没有。我已经完成了这一天，但如果您可以将您的 Transaction 课程添加到问题中，我明天可以看看。
啊 - 可怕的龙目岛 - 但我看到你有一个由它创建的无参数 CTOR - 我不知道它为什么要寻找需要 int 的 CTOR - 特别是考虑到它没有 int字段；虽然它确实有一个数字 - 也许这是 MessagePack 的一些工件。添加public Transaction(int dummy) { } 看看会发生什么不会有什么坏处。
由于您在@Bean 方法中显式实例化Serde 对象，您可以使用构造函数new JsonSerde<>(Transaction.class, mapper)，从而避免上面提到的属性Gary（default.type）。但是，正如 Gary 所说，我认为您最近的错误是由于您使用的消息包库造成的。
啊，我发现了错误。在我发布到transactions 主题的python 程序中，我发送的是msgpack.packb(json.dumps({"item": "private jet", amount: 240})) 而不仅仅是msgpack.packb({"item": "private jet", amount: 240})。感谢你们的帮助，也感谢你们在 Spring Cloud Stream 上的出色工作。到目前为止，我真的很喜欢使用它，并且从你的 GitHub 示例中学到了很多东西。