深入理解Apache Kafka(分布式流处理平台)

<dependency>
    <groupid>org.apache.kafka</groupid>
    <artifactid>kafka-clients</artifactid>
    <version>3.4.0</version>
</dependency>

import org.apache.kafka.clients.producer.*;
import java.util.properties;
public class kafkaproducerexample {
    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);
        // 发送消息
        for (int i = 0; i < 10; i++) {
            producerrecord<string, string> record = new producerrecord<>(
                "test-topic", 
                "key-" + i, 
                "message-" + i
            );
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    exception.printstacktrace();
                } else {
                    system.out.printf("message sent to partition %d with offset %d%n",
                            metadata.partition(), metadata.offset());
                }
            });
        }
        // 关闭生产者
        producer.close();
    }
}

import org.apache.kafka.clients.consumer.*;
import java.time.duration;
import java.util.collections;
import java.util.properties;
public class kafkaconsumerexample {
    public static void main(string[] args) {
        // 配置消费者属性
        properties props = new properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.stringdeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.stringdeserializer");
        // 创建消费者实例
        consumer<string, string> consumer = new kafkaconsumer<>(props);
        // 订阅topic
        consumer.subscribe(collections.singletonlist("test-topic"));
        // 轮询获取消息
        try {
            while (true) {
                consumerrecords<string, string> records = consumer.poll(duration.ofmillis(100));
                for (consumerrecord<string, string> record : records) {
                    system.out.printf("received message: key = %s, value = %s, partition = %d, offset = %d%n",
                            record.key(), record.value(), record.partition(), record.offset());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

// 简单的流处理示例
streamsbuilder builder = new streamsbuilder();
builder.stream("input-topic")
       .mapvalues(value -> value.tostring().touppercase())
       .to("output-topic");
kafkastreams streams = new kafkastreams(builder.build(), props);
streams.start();