在 Spring、Kafka 和 Kotlin 中正确使用 Avro

Question

通过我的在线申请：

• 用户在浏览器中单击一个按钮，该按钮对应用程序进行安静的调用
• 控制器接收到 RESTful 调用并构建命令并接受拒绝命令
• 如果有效，则将命令触发到 Kafka 主题
• 应用程序代码使用命令并将命令提供给适当的处理程序或处理程序
• 处理程序对域概念执行命令
• 域概念会将事件触发到 Kafka 主题

• 再次接收消息并转发给感兴趣的处理程序的应用程序消费者

• 示例命令

  • RegisterClientCommand
  • PlaceBookingCommand
  • CancelBookingCommand

• 示例事件

  • ClientRegisteredEvent
  • BookingPlacedEvent
  • BookingCancelledEvent

• 命令和事件的类型都是DomainMessage

• 每个域消息都有一个sequence-number、createdOn 字段

• 我决定使用具有模式版本支持的 Avro 来进行消息交换

• 春季卡夫卡制作人：

  • ProducerFactory<String?, GenericRecord?>

• 春季卡夫卡消费者：

  • ConsumerFactory<String?, GenericRecord?>

发送 Kafka 消息时的点 (1)，GenericRecord 的创建方式如下：

return GenericRecordBuilder(schema).apply {
    set("first", registerClientCommand.first)
    set("last", registerClientCommand.last)
    set("email", registerClientCommand.email)
    set("mobile", registerClientCommand.mobile)
  }.build()

收到消息时的点(2)： - 从 GenericRecord 构建地图 - 并使用 Map 中的字段值创建一个具体的 RegisterClientCommand

return RegisterClientCommand(
    fields.get("first"),
    fields.get("last"),
    fields.get("email"),
    fields.get("mobile")
  )

你觉得这正确吗？

我不喜欢我必须：对于每个 DomainMessage（命令、事件）

• 在生产到 Kafka 时构建 GenericRecord，如上述第 (1) 点所述。
• 并构建一个字段值映射，然后构建具体的域消息 从 Kafka 消费时，如上述第 (2) 点所述。

我在这里以正确的方式使用 Avro 吗？

Answer 1

使用 Spring Kafka 库怎么样？这可以处理各种转换和订阅者。你需要提供一个RecordMessageConverter的bean，这个bean必须实现简单的Avro转换，将POJO转换为Avro，从Avro消息转换为POJO。

消息推送也可以简化，需要提供beanTopic

消息发送看起来像

@Service
public class Kafka {
    @Autowired
    private KafkaTemplate<Object, Object> template;
    public void sendClass1(Class1 object) {
        this.template.send("topic1", object);
    }

}