向apache kafka发送tomcat服务器日志和指标的方法

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

Properties props = new Properties();
props.put("bootstrap.servers", "kafka-server1:9092,kafka-server2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

String topic = "logs";

// 发送日志消息
String logMessage = "Your log message";
producer.send(new ProducerRecord<>(topic, logMessage));

// 发送指标数据
String metricData = "Your metric data";
producer.send(new ProducerRecord<>(topic, metricData));

producer.close();

bin/kafka-topics.sh --create --bootstrap-server kafka-server:9092 --replication-factor 1 --partitions 1 --topic logs

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

Properties props = new Properties();
props.put("bootstrap.servers", "kafka-server1:9092,kafka-server2:9092");
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);

String topic = "logs";
consumer.subscribe(Collections.singletonList(topic));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    // 在这里处理接收到的日志和指标消息
    
    consumer.commitAsync();
}

consumer.close();