springboot 1.5.2 集成kafka的简单例子

首先我们需要明确，使用Spring Boot集成Kafka主要需要以下几个步骤：

1. 导入kafka相关的依赖
2. 配置Kafka连接参数
3. 实现生产者和消费者

下面我们来一步一步进行操作。

导入kafka相关的依赖

在Spring Boot项目中，我们可以使用Maven或Gradle来管理依赖。在这里我们选择使用Maven，我们在项目的pom.xml文件中加入以下依赖：

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>1.3.9.RELEASE</version>
</dependency>
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.1.0</version>
</dependency>

配置Kafka连接参数

在Spring Boot中，我们可以使用配置文件来配置Kafka连接参数。在application.properties中加入以下配置：

spring.kafka.producer.bootstrap-servers=localhost:9092
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=test-group

实现生产者

下面我们来看一个简单的生产者示例：

@Service
public class KafkaProducer {

  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;

  public void sendMessage(String topic, String message) {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

      @Override
      public void onSuccess(SendResult<String, String> result) {
        System.out.println("Sent message=[" + message + "] with offset=[" + result.getRecordMetadata().offset() + "]");
      }

      @Override
      public void onFailure(Throwable ex) {
        System.out.println("Unable to send message=[" + message + "] due to : " + ex.getMessage());
      }
    });
  }
}

在这个示例中，我们使用KafkaTemplate来发送Kafka消息。其中，topic表示消息要发送到哪个主题，message表示消息的内容。

实现消费者

下面我们来看一个简单的消费者示例：

@Service
public class KafkaConsumer {

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

在这个示例中，我们使用@KafkaListener注解来监听Kafka主题。其中，topics表示要监听的主题，groupId表示消费者所属的组。

验证

最后，我们来验证一下我们的Kafka集成是否成功。我们可以在控制台上启动一个Kafka消费者，指定监听的主题为test-topic和groupId为test-group，从而接收到生产者发送的消息。

另外，我们也可以编写一个简单的测试方法，来验证生产者是否能够成功发送消息：

@Autowired
private KafkaProducer kafkaProducer;

@Test
public void sendMessageTest() {
  kafkaProducer.sendMessage("test-topic", "Hello, Kafka!");
}

至此，我们完成了Spring Boot集成Kafka的简单例子。