SpringBoot整合SpringKafka实现消费者史上最简代码实现

该项目是使用的技术：SpringBoot + SpringKafka + Maven

先看pom.xml文件中引入的依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>

<groupId>com.xuebusi.consumer</groupId>
<artifactId>consumer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>springkafkaconsumer</name>
<description>Demo project for Spring Boot</description>

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.8.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-freemarker</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>1.0.6.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>

</project>

注意：这里我使用的spring-kafka（它包装了apache的kafka-client）的依赖包版本是 1.0.6.RELEASE， 是因为我Linux服务器上部署的kafka服务器的版本是kafka_2.10-0.9.0.1，使用的kafka的时候要注意，kafka客户端（kafka-client）的版本要和kafka服务器的版本一一对应，否则，消息会消费失败。

Spring官方网站上给出了SpringKafka和kafka-client版本（它的版本号要和kafka服务器的版本保持一致）的对应关系：
https://projects.spring.io/spring-kafka/


下面是生产者的配置文件，既然使用的是SpringBoot，配置文件就是 application.yml：

server:
port: 8082
spring:
kafka:
consumer:
enable-auto-commit: true
group-id: applog
auto-offset-reset: latest
bootstrap-servers: 192.168.71.11:9092,192.168.71.12:9092,192.168.71.13:9092

在上面的配置中，我们给消费者分配的端口号是8082，服务器有3台，分别对应3个ip地址和端口。 并配置了kafka服务器的ip地址；

enable-auto-commit: true //指定消息被消费之后自动提交偏移量（即消息的编号，表示消费到了哪个位置，消费者每消费完一条消息就会向kafka服务器汇报自己消消费到的那个消息的编号，以便于下次继续消费）。
group-id: applog //消费者组
auto-offset-reset: latest //从最近的地方开始消费

想了解关于kafka消费者相关的更多配置的话，可以自行查阅相关资料进行学习。

下面是kafka消费者的核心代码，实现了消息的消费逻辑：

package com.xuebusi.consumer;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
* 消费者
* 使用@KafkaListener注解,可以指定:主题,分区,消费组
*/
@Component
public class KafkaConsumer {

@KafkaListener(topics = {"app_log"})
public void receive(String message){
System.out.println("app_log--消费消息:" + message);
}
}

在上面的代码中，负责消费消息的关键之处就是SpringKafka提供的@KafkaListener注解，在方法上使用该注解，并指定要消费的topic（也可以指定消费组以及分区号，支持正则表达式匹配），这样，消费者一旦启动，就会监听kafka服务器上的topic，实时进行消费消息。当然，我们可以在该类中定义多个不同的方法，并都在方法上使用 @KafkaListener ，为它指定不同的topic及分区信息，这样每个方法就相当于一个消费者了。

下面就启动SpringBoot项目测试：

package com.xuebusi;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringkafkaconsumerApplication {

public static void main(String[] args) {
SpringApplication.run(SpringkafkaconsumerApplication.class, args);
}
}

使用鼠标右键运行main方法即可启动SpringBoot项目，在控制台你会看到消费者会成功连接到kafka服务器，并打印出消费者的参数配置信息以及消费者和topic分区的分配信息，消费者一旦发现要消费的topic中有新的消息，就会立即进行消费。