SpringBoot整合SpringKafka實現生產者史上最簡代碼實現

項目源碼:https://github.com/xuebus/springkafkaproducer

這是生產者端的代碼,消費者請參考另一篇博客:

https://www.cnblogs.com/jun1019/p/7895830.html

該項目是使用的技術：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.producer</groupId>
    <artifactId>producer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>springkafkaproducer</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: 8081 spring:
  kafka:
    producer:
      bootstrap-servers: 192.168.71.11:9092,192.168.71.12:9092,192.168.71.13:9092

在上面的配置中，我們給生產者分配的端口號是8081，服務器有3台，分別對應3個ip地址和端口。
該配置只是配置了kafka服務器的ip地址，這里並沒有對生產者做過多的配置。想了解關於kafka生產者相關的更多配置的話，可以自行查閱相關資料進行學習。

 

下面是kafka生產者的核心代碼，實現了消息的發送邏輯：

package com.xuebusi.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.UUID;

/**
 * 生產者
 * 使用@EnableScheduling注解開啟定時任務
 */
@Component
@EnableScheduling
public class KafkaProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 定時任務
     */
    @Scheduled(cron = "00/1 * * * * ?")
    public void send(){
        String message = UUID.randomUUID().toString();
        ListenableFuture future = kafkaTemplate.send("app_log", message);
        future.addCallback(o -> System.out.println("send-消息發送成功：" + message), throwable -> System.out.println("消息發送失敗：" + message));
    }

}

在上面的代碼中，負責發送消息的角色就是SpringKafka提供的 KafkaTemplate對象，使用它的 send()方法就可以把自己的消息發送到kafka服務器。send()方法有多個重載的方法，大家可以根據自己的需要來使用不同的send()方法。

這里我們為了方便發送消息進行測試，使用了Spring的定時任務，在類上使用 @EnableScheduling 注解開啟定時任務，在方法上使用@Scheduled注解並指定表達式來定義定時規則，這里我們每秒就會向kafka服務器發送一條消息。

當然，你可能不會用到Spring的定時任務，你可以把@EnableScheduling  和 @Scheduled注解去掉。你也可以通過使用Spring的@Controller和@RequestMapping 注解將你的發送消息的方法定義成一個接口。

你可以定義多個不同的方法，每個使用到了 KafkaTemplate 對象的方法都會是一個生產者。

 

下面就啟動SpringBoot項目測試：

package com.xuebusi;

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

@SpringBootApplication
public class SpringkafkaproducerApplication {

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

使用鼠標右鍵運行main方法即可啟動SpringBoot項目，在控制台你會看到成功連接到kafka服務器，並每隔1秒就會發送一條消息到kafka服務器。