Kafka

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");
        props.put("acks", "all");  // 等待所有副本确认
        props.put("retries", 3);   // 重试次数

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

        try {
            // 同步发送
            ProducerRecord<String, String> record =
                new ProducerRecord<>("my-topic", "key1", "value1");
            RecordMetadata metadata = producer.send(record).get();
            System.out.printf("Sent to partition %d with offset %d%n",
                metadata.partition(), metadata.offset());

            // 异步发送
            producer.send(new ProducerRecord<>("my-topic", "key2", "value2"),
                new Callback() {
                    public void onCompletion(RecordMetadata metadata, Exception e) {
                        if (e != null) {
                            e.printStackTrace();
                        } else {
                            System.out.printf("Sent to partition %d%n", metadata.partition());
                        }
                    }
                });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "earliest"); // earliest, latest, none

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("my-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                        record.topic(), record.partition(), record.offset(),
                        record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.util.Properties;

public class WordCountExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();

        // 读取输入 Topic
        KStream<String, String> textLines = builder.stream("input-topic");

        // 处理流
        KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count();

        // 写入输出 Topic
        wordCounts.toStream().to("output-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

// 使用幂等性 Producer
props.put("enable.idempotence", true);

// 或使用事务
producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// 幂等性 Producer
Properties props = new Properties();
props.put("enable.idempotence", true);

// 事务性 Producer
producer.initTransactions();
producer.beginTransaction();
// ... send messages
producer.commitTransaction();

// Producer: 收集应用日志
public class LogCollector {
    private final KafkaProducer<String, String> producer;

    public void sendLog(String app, String log) {
        ProducerRecord<String, String> record =
            new ProducerRecord<>("app-logs", app, log);
        producer.send(record);
    }
}

// Consumer: 存储到 Elasticsearch
public class LogStorage {
    public void process(ConsumerRecords<String, String> records) {
        for (ConsumerRecord<String, String> record : records) {
            elasticsearchClient.index(record.value());
        }
    }
}

// Kafka Streams: 实时聚合
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Transaction> transactions = builder.stream("transactions");

KTable<String, Double> revenue = transactions
    .groupBy((key, tx) -> tx.getUserId())
    .aggregate(
        () -> 0.0,
        (key, tx, sum) -> sum + tx.getAmount(),
        Materialized.as("revenue-store")
    );

revenue.toStream().to("user-revenue");

// 保存事件
public class EventStore {
    public void saveEvent(DomainEvent event) {
        ProducerRecord<String, DomainEvent> record =
            new ProducerRecord<>("domain-events", event.getAggregateId(), event);
        producer.send(record);
    }
}

// 重建状态
public class StateRebuilder {
    public AggregateState rebuild(String aggregateId) {
        consumer.subscribe(Collections.singletonList("domain-events"));
        AggregateState state = new AggregateState();

        while (true) {
            ConsumerRecords<String, DomainEvent> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, DomainEvent> record : records) {
                if (record.key().equals(aggregateId)) {
                    state.apply(record.value());
                }
            }
        }
    }
}