Spring-Boot集成Kafka

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Service
public class KafkaProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    // 构造器注入KafkaTemplate
    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 同步发送消息（阻塞等待结果）
     */
    public void sendMessageSync(String topic, String key, String message) {
        try {
            // 发送消息并等待确认
            SendResult<String, String> result = kafkaTemplate.send(topic, key, message).get();
            System.out.println("发送成功 -> Topic: " + topic 
                + ", Partition: " + result.getRecordMetadata().partition() 
                + ", Offset: " + result.getRecordMetadata().offset());
        } catch (Exception e) {
            System.err.println("发送失败: " + e.getMessage());
            // 可添加重试或补偿逻辑
        }
    }

    /**
     * 异步发送消息（非阻塞，回调处理）
     */
    public void sendMessageAsync(String topic, String key, String message) {
        ListenableFuture<SendResult<String, String>> future = 
            kafkaTemplate.send(topic, key, message);

        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("发送成功 -> Topic: " + topic 
                    + ", Partition: " + result.getRecordMetadata().partition());
            }

            @Override
            public void onFailure(Throwable ex) {
                System.err.println("发送失败: " + ex.getMessage());
                // 可记录失败消息到数据库或重试队列
            }
        });
    }
}

public void sendToPartition(String topic, int partition, String key, String message) {
    // 直接使用带分区参数的 send 方法
    kafkaTemplate.send(topic, partition, key, message)
        .addCallback(
            result -> handleSuccess(result),
            ex -> handleFailure(ex)
        );
}

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, 
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        
        // 示例：根据键的哈希值计算分区
        if (key != null) {
            return Math.abs(key.hashCode()) % numPartitions;
        }
        
        // 没有键时使用随机分区
        return ThreadLocalRandom.current().nextInt(numPartitions);
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    // 监听指定Topic，并发消费者线程数为3（根据分区数调整）
    @KafkaListener(
            topics = "your-topic-name",
            concurrency = "3",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void consume(String message, Acknowledgment ack) {
        try {
            // 1. 处理消息逻辑
            System.out.println("Received message: " + message);
            // 模拟业务处理（如数据库操作、计算等）
            processMessage(message);

            // 2. 手动提交偏移量（确保消息处理成功后再提交）
            ack.acknowledge();
        } catch (Exception e) {
            // 3. 处理异常（如重试、记录日志等）
            System.err.println("消费失败: " + e.getMessage());
            // 可根据业务决定是否提交偏移量（如不提交则下次重试）
        }
    }

    private void processMessage(String message) {
        // 实际业务逻辑
    }
}

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        // 其他配置...

        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public KafkaTransactionManager<String, String> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(producerFactory());
    }
}

@Service
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendTransactionalMessage() {
        kafkaTemplate.executeInTransaction(operations -> {
            try {
                operations.send("topic1", "key1", "value1");
                operations.send("topic2", "key2", "value2");
                // 其他业务逻辑...
                // 如果一切正常，事务会自动提交
            } catch (Exception e) {
                // 如果发生异常，事务会自动回滚
                throw e;
            }
        });
    }
}