Spring纯Java配置集成kafka代码实例

下面我将详细讲解如何使用Spring纯Java配置集成kafka，包括以下步骤：

1. 添加依赖
2. 配置Kafka
3. 发送消息
4. 接收消息

1. 添加依赖

首先，我们需要在项目的pom.xml中添加kafka相关的依赖，如下所示：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.7.3</version>
</dependency>

2. 配置Kafka

要使用Kafka，我们需要配置一个ProducerFactory和一个KafkaTemplate。我们可以使用Java Config方式进行配置，如下所示：

@Configuration
@EnableKafka
public class KafkaConfig {

    @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);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

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

在这个配置类中，我们创建了一个ProducerFactory，它使用Kafka的默认配置来创建生产者。然后我们创建了一个KafkaTemplate，并将其注入到Spring容器中。

3. 发送消息

接下来，我们可以使用注入的KafkaTemplate来发送消息到Kafka。例如，下面的代码展示了如何发送字符串消息：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String message) {
    kafkaTemplate.send("topic-name", message);
}

4. 接收消息

要接收从Kafka发送的消息，我们需要创建一个消费者，并订阅一个或多个主题来接收消息。我们可以使用Java Config方式来配置消费者，如下所示：

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @KafkaListener(topics = "topic-name", groupId = "group-id")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}

在这个配置类中，我们创建了一个ConsumerFactory和一个KafkaListenerContainerFactory。然后我们使用@KafkaListener注解来创建一个监听器，它将订阅“topic-name”主题来接收消息。

例如，下面的代码展示了如何在Spring Boot应用程序中使用Kafka发送和接收消息并打印到控制台：

@RestController
public class KafkaController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send-message")
    public void sendMessage(@RequestParam String message) {
        kafkaTemplate.send("topic-name", message);
    }

    @KafkaListener(topics = "topic-name", groupId = "group-id")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}

以上就是使用Spring纯Java配置集成kafka的完整攻略，其中包含了发送和接收消息的两条示例。