Spring Boot+Maven下基于AsyncAPI YAML生成POJO及实现Kafka生产者

一、加载asyncApi.yaml文件到Java代码 #

1. 放置文件：将asyncApi.yaml放在Maven项目的src/main/resources目录下，这是Spring默认的资源加载路径。
2. 引入依赖：添加AsyncAPI Java解析器的Maven依赖，用于解析YAML内容：

   <dependency>
       <groupId>com.asyncapi</groupId>
       <artifactId>asyncapi-java-spring-boot-starter</artifactId>
       <version>3.0.0</version> <!-- 建议使用最新稳定版 -->
   </dependency>
   

3. 代码加载：通过Spring的资源加载类读取并解析文件：

   import org.springframework.core.io.ClassPathResource;
   import com.asyncapi.v3.parser.AsyncAPIParser;
   import com.asyncapi.v3.model.AsyncAPI;
   import java.io.IOException;
   
   // 在Spring Bean中加载AsyncAPI文件
   public AsyncAPI loadAsyncApiSpec() throws IOException {
       ClassPathResource resource = new ClassPathResource("asyncApi.yaml");
       return new AsyncAPIParser().parse(resource.getInputStream());
   }
   

二、基于AsyncAPI配置实现Kafka生产者 #

你可以从解析后的AsyncAPI对象中提取Kafka配置、主题和消息结构，结合Spring Kafka实现生产者：

1. 提取Kafka服务器与安全配置（可选，用于动态覆盖application.properties）：

   AsyncAPI asyncAPI = loadAsyncApiSpec();
   // 获取服务器地址（需对应yaml中定义的server名称）
   String bootstrapServers = asyncAPI.getServers().get("kafka-prod").getUrl();
   // 获取SASL安全配置
   Map<String, Object> securityConfig = asyncAPI.getServers().get("kafka-prod")
           .getSecurity().get(0).get("sasl").getProperties();
   

2. 获取主题与消息定义：

   // 获取要发送的主题（示例取第一个定义的通道）
   String topic = asyncAPI.getChannels().keySet().stream().findFirst().orElse("user-events");
   // 获取消息的Schema定义，用于校验消息结构
   com.asyncapi.v3.model.schema.Schema messageSchema = asyncAPI.getChannels().get(topic)
           .getPublish().getMessage().getPayload();
   

3. 结合KafkaTemplate发送消息：
   直接使用Spring Kafka的KafkaTemplate，发送的消息体可以用后面生成的POJO，确保符合AsyncAPI定义的结构：

   @Autowired
   private KafkaTemplate<String, UserEvent> kafkaTemplate;
   
   public void sendEvent(UserEvent event) {
       kafkaTemplate.send(topic, event);
   }
   

三、基于asyncApi.yaml自动生成POJO类 #

使用AsyncAPI Maven插件可以自动根据YAML中的消息Schema生成Java实体类：

1. 添加Maven插件配置：在pom.xml的<build><plugins>节点下添加：

   <plugin>
       <groupId>com.asyncapi</groupId>
       <artifactId>asyncapi-maven-plugin</artifactId>
       <version>3.0.0</version>
       <executions>
           <execution>
               <goals>
                   <goal>generate</goal>
               </goals>
               <configuration>
                   <inputSpec>src/main/resources/asyncApi.yaml</inputSpec>
                   <generator>java-spring</generator> <!-- 指定Spring风格的生成器 -->
                   <output>src/main/java</output> <!-- 生成代码的目标目录 -->
                   <config>
                       <packageName>com.yourproject.kafka.model</packageName> <!-- 自定义包路径 -->
                       <useJackson>true</useJackson> <!-- 生成Jackson序列化注解 -->
                   </config>
               </configuration>
           </execution>
       </executions>
   </plugin>
   

2. 执行生成命令：在项目根目录运行Maven命令：

   mvn asyncapi:generate
   

   插件会自动根据YAML中定义的消息payload生成对应的POJO，包含getter/setter、序列化注解，可直接用于Kafka消息的发送与接收。

注意事项 #

• 确保asyncApi.yaml中的Kafka配置（服务器地址、安全机制）与application.properties保持一致，避免冲突。
• AsyncAPI Java解析器和Maven插件建议使用相同版本，避免兼容性问题。
• 生成POJO时，可通过插件的config节点自定义更多参数，比如是否生成Builder模式、是否添加Lombok注解等。

内容的提问来源于stack exchange，提问作者blaiso