使用 spring+kafka demo

以下是使用Spring和Kafka的示例代码：

添加 Maven 依赖项

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

配置 Kafka

@Configuration
@EnableKafka
public class KafkaConfig {

  @Value("${spring.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, "my-group-id");
    return props;
  }

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

  @Bean
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(3);
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
  }

  @Bean
  public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return props;
  }

  @Bean
  public ProducerFactory<String, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
  }

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

}

发送和接收 Kafka 消息

@RestController
public class KafkaController {

  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;

  @GetMapping("/send")
  public void sendMessage() {
    kafkaTemplate.send("my-topic", "Hello, Kafka!");
  }

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

}

在上面的示例中，我们首先配置了 Kafka 的消费者和生产者。然后我们创建了一个 KafkaController 类，其中包含一个发送 Kafka 消息的方法和一个接收 Kafka 消息的方法。我们使用 KafkaTemplate 来发送消息，并使用 @KafkaListener 注解来监听消息。当接收到消息时，我们使用 System.out.println() 方法将其打印到控制台上。

总的来说，使用 Spring 和 Kafka 的过程非常简单。只要你理解了 Kafka 的基本概念，就可以轻松地将其与 Spring 集成。