Apache Kafka 技术文档

// 1. 同步发送
producer.send(record).get();

// 2. 异步发送
producer.send(record, new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println("消息发送成功: " + metadata.offset());
        }
    }
});

// 3. 批量发送
producer.send(record1);
producer.send(record2);
producer.flush(); // 确保所有消息发送完成

// 1. 轮询策略（默认）
// 消息会轮询分配到各个分区

// 2. 随机策略
// 随机选择分区

// 3. 按键分区策略
// 根据消息键的哈希值选择分区
producer.send(new ProducerRecord<>("my-topic", "key", "value"));

// 4. 自定义分区策略
public class CustomPartitioner implements Partitioner {
    public int partition(String topic, Object key, byte[] keyBytes,
                       Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义分区逻辑
        return Math.abs(key.hashCode()) % cluster.partitionCountForTopic(topic);
    }
}

// 1. 自动提交偏移量
Properties props = new Properties();
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");

// 2. 手动提交偏移量
props.put("enable.auto.commit", "false");
// 在消息处理完成后手动提交
consumer.commitSync();

// 3. 异步提交偏移量
consumer.commitAsync(new OffsetCommitCallback() {
    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
                          Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        }
    }
});

// 1. 指定偏移量开始消费
consumer.seek(partition, offset);

// 2. 从最早位置开始消费
consumer.seekToBeginning(partitions);

// 3. 从最新位置开始消费
consumer.seekToEnd(partitions);

// 4. 暂停和恢复消费
consumer.pause(partitions);
consumer.resume(partitions);

// 1. 轮询分区（默认）
// 消息会轮询分配到各个分区

// 2. 按键分区
// 相同键的消息会发送到同一分区
producer.send(new ProducerRecord<>("topic", "key", "value"));

// 3. 指定分区
// 直接指定目标分区
producer.send(new ProducerRecord<>("topic", 0, "key", "value"));

// 1. 轮询分区（默认）
// 消息会轮询分配到各个分区

// 2. 按键分区
// 相同键的消息会发送到同一分区
producer.send(new ProducerRecord<>("topic", "key", "value"));

// 3. 指定分区
// 直接指定目标分区
producer.send(new ProducerRecord<>("topic", 0, "key", "value"));

// 1. 自动提交偏移量
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");

// 2. 手动提交偏移量
props.put("enable.auto.commit", "false");
// 在消息处理完成后手动提交
consumer.commitSync();

// 3. 指定偏移量开始消费
consumer.seek(partition, offset);

// 1. 字符串序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 2. JSON 序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 3. Avro 序列化
props.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

// 1. 创建 Producer
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);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);

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

// 2. 发送消息
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
producer.send(record);

// 3. 关闭 Producer
producer.close();

// 1. 异步发送回调
producer.send(record, new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println("消息发送成功: " + metadata.offset());
        }
    }
});

// 2. 批量发送
for (int i = 0; i < 100; i++) {
    ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key" + i, "value" + i);
    producer.send(record);
}
producer.flush(); // 确保所有消息发送完成

// 3. 事务发送
producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// 1. 创建 Consumer
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("auto.offset.reset", "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 2. 订阅 Topic
consumer.subscribe(Arrays.asList("my-topic"));

// 3. 消费消息
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("消费消息: " + record.value());
    }
}

// 1. 手动提交偏移量
props.put("enable.auto.commit", "false");
consumer.commitSync();

// 2. 指定