Finology 大数据金融

@SpringBootApplication(exclude = {
		KafkaAutoConfiguration.class
})

package gy.finolo.config.kafka;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;


@Configuration
public class KafkaConfig {

        @Bean
        @ConfigurationProperties(prefix = "spring.kafka.cluster1")
        public KafkaProperties cluster1KafkaProperties() {
                return new KafkaProperties();
        }

        @Bean
        @ConfigurationProperties(prefix = "spring.kafka.cluster2")
        public KafkaProperties cluster2KafkaProperties() {
                return new KafkaProperties();
        }

        @Bean("cluster1KafkaTemplate")
        public KafkaTemplate<String, String> cluster1KafkaTemplate() {
                return new KafkaTemplate<>(cluster1ProducerFactory());
        }

        @Bean("cluster2KafkaTemplate")
        public KafkaTemplate<String, String> cluster2KafkaTemplate() {
                return new KafkaTemplate<>(cluster2ProducerFactory());
        }

        @Bean("cluster1ProducerFactory")
        public ProducerFactory<String, String> cluster1ProducerFactory() {
                // 直接使用KafkaProperties，更简洁
                Map<String, Object> configProps = new HashMap<>();

                // 使用KafkaProperties的getter方法
                configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                                cluster1KafkaProperties().getBootstrapServers());
                configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                                cluster1KafkaProperties().getProducer().getKeySerializer());
                configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                                cluster1KafkaProperties().getProducer().getValueSerializer());

                // 添加其他必要的配置
                configProps.put(ProducerConfig.ACKS_CONFIG,
                                cluster1KafkaProperties().getProducer().getAcks());
                configProps.put(ProducerConfig.RETRIES_CONFIG,
                                cluster1KafkaProperties().getProducer().getRetries());

                return new DefaultKafkaProducerFactory<>(configProps);
        }

        @Bean("cluster2ProducerFactory")
        public ProducerFactory<String, String> cluster2ProducerFactory() {
                Map<String, Object> configProps = new HashMap<>();

                // 使用KafkaProperties的getter方法
                configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                                cluster2KafkaProperties().getBootstrapServers());
                configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                                cluster2KafkaProperties().getProducer().getKeySerializer());
                configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                                cluster2KafkaProperties().getProducer().getValueSerializer());

                // 添加其他必要的配置
                configProps.put(ProducerConfig.ACKS_CONFIG,
                                cluster2KafkaProperties().getProducer().getAcks());
                configProps.put(ProducerConfig.RETRIES_CONFIG,
                                cluster2KafkaProperties().getProducer().getRetries());

                return new DefaultKafkaProducerFactory<>(configProps);
        }

}

@Slf4j
@Service
public class KafkaMessageService {

    @Autowired
    @Qualifier("cluster1KafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;

    public CompletableFuture<SendResult<String, String>> sendMessage(String message) {
        String topic = "kafka_topic";
        return kafkaTemplate.send(topic, message);
    }
}