如何在Java TestServlet中获取Postman传递的JSON参数并写入Kafka

1. 依赖准备 #

在Java项目中添加以下依赖（以Maven为例）：

• Servlet API（若Servlet容器自带可省略）
• Kafka Java客户端
• Jackson JSON解析库（用于处理请求体的JSON数据）

<dependencies>
    <!-- Servlet API -->
    <dependency>
        <groupId>javax.servlet</groupId>
        <artifactId>javax.servlet-api</artifactId>
        <version>4.0.1</version>
        <scope>provided</scope>
    </dependency>
    <!-- Kafka Client -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.6.1</version>
    </dependency>
    <!-- Jackson JSON -->
    <dependency>
        <groupId>com.fasterxml.jackson.databind</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.15.2</version>
    </dependency>
</dependencies>

2. 编写TestServlet处理POST请求 #

重写doPost方法，读取并解析请求体中的JSON参数，调用Kafka Producer发送消息到指定Topic：

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.io.BufferedReader;
import java.io.IOException;
import java.util.Properties;

@WebServlet("/xyz/TestServlet")
public class TestServlet extends HttpServlet {
    private KafkaProducer<String, String> kafkaProducer;
    private final ObjectMapper objectMapper = new ObjectMapper();
    // 目标Kafka Topic名称
    private static final String KAFKA_TOPIC = "your-target-topic";

    @Override
    public void init() throws ServletException {
        // 初始化Kafka Producer配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-ip:9092"); // 替换为你的Kafka Broker地址
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 可选配置：确保消息可靠性
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);

        kafkaProducer = new KafkaProducer<>(props);
    }

    @Override
    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        response.setContentType("application/json");
        response.setCharacterEncoding("UTF-8");

        // 读取请求体中的JSON数据
        StringBuilder requestBody = new StringBuilder();
        String line;
        BufferedReader reader = request.getReader();
        while ((line = reader.readLine()) != null) {
            requestBody.append(line);
        }

        try {
            // 解析JSON参数
            JsonNode jsonNode = objectMapper.readTree(requestBody.toString());
            String firstname = jsonNode.get("firstname").asText();
            String lastname = jsonNode.get("lastname").asText();
            String email = jsonNode.get("email").asText();
            String phone = jsonNode.get("phone").asText();

            // 将参数封装为JSON字符串作为Kafka消息体
            String message = objectMapper.writeValueAsString(jsonNode);

            // 发送消息到Kafka Topic
            ProducerRecord<String, String> record = new ProducerRecord<>(KAFKA_TOPIC, email, message); // 用email作为消息key，可选
            kafkaProducer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                    response.getWriter().write("{\"status\":\"error\",\"message\":\"Kafka消息发送失败\"}");
                } else {
                    response.getWriter().write("{\"status\":\"success\",\"message\":\"消息已发送至Kafka\"}");
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
            response.getWriter().write("{\"status\":\"error\",\"message\":\"JSON参数解析失败\"}");
        }
    }

    @Override
    public void destroy() {
        // 关闭Kafka Producer，释放资源
        if (kafkaProducer != null) {
            kafkaProducer.close();
        }
    }
}

3. 关键注意事项 #

• Kafka Broker地址：替换代码中kafka-broker-ip:9092为你的实际Kafka集群地址
• Topic配置：修改KAFKA_TOPIC为目标Topic名称，需确保Topic已提前创建，或开启Kafka自动创建Topic的配置
• 请求格式：Postman发送请求时必须设置Content-Type为application/json，否则Servlet无法正确解析请求体
• 性能优化：通过Servlet的init方法初始化Kafka Producer单例，避免每次请求重复创建实例
• 异常处理：代码包含基础异常捕获，生产环境可根据需求优化日志记录和错误返回逻辑

4. Postman测试步骤 #

1. 选择POST请求，填入URL：https://xx.xxx.xxx.xxxx:yyyy/xyz/TestServlet
2. 在Headers标签添加：Key: Content-Type, Value: application/json
3. 在Body标签选择raw，格式选JSON，输入测试参数：

{
    "firstname": "John",
    "lastname": "Doe",
    "email": "john.doe@example.com",
    "phone": "1234567890"
}

4. 点击Send查看响应结果，同时可通过Kafka消费者验证消息是否成功写入目标Topic

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