Wie gehe ich mit der asynchronen Kommunikation und Verarbeitung von Formulardaten in der Nachrichtenwarteschlange in Java um?

// HTML页面中的表单
<form id="myForm">
    <input type="text" name="username">
    <input type="password" name="password">
    <input type="submit" value="Submit" onclick="submitForm()">
</form>

// JavaScript代码
function submitForm() {
    var form = document.getElementById("myForm");
    var xhr = new XMLHttpRequest();
    xhr.open("POST", "MyServlet", true);
    xhr.setRequestHeader("Content-Type", "application/x-www-form-urlencoded");
    xhr.onreadystatechange = function() {
        if (xhr.readyState === XMLHttpRequest.DONE && xhr.status === 200) {
            console.log(xhr.responseText);
        }
    };
    xhr.send(new FormData(form));
}

// Servlet代码
@WebServlet("/MyServlet")
public class MyServlet extends HttpServlet {
    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        String username = request.getParameter("username");
        String password = request.getParameter("password");
        // 处理表单数据...
        response.getWriter().write("Data received and processed successfully!");
    }
}

// Producer代码
public class FormProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        Scanner scanner = new Scanner(System.in);

        System.out.println("Please enter your username:");
        String username = scanner.nextLine();
        System.out.println("Please enter your password:");
        String password = scanner.nextLine();

        producer.send(new ProducerRecord<>("form-data", username + "," + password));
        producer.close();
    }
}

// Consumer代码
public class FormConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "form-consumer-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("form-data"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                String[] data = record.value().split(",");
                // 处理表单数据...
                System.out.println("Data received and processed successfully!");
            }
        }
    }
}