Spring boot 集成Kafka

搭建Kafka集群，参考：

https://www.cnblogs.com/jonban/p/kafka.html

源码示例如下：

1、新建 Maven 项目 kafka

2、pom.xml

<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.java</groupId>

    <artifactId>kafka</artifactId>

    <version>1.0.0</version>

    <parent>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-parent</artifactId>

        <version>2.0.5.RELEASE</version>

    </parent>

    <dependencies>

        <!-- Spring Boot -->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-web</artifactId>

        </dependency>

        <dependency>

            <groupId>org.springframework.kafka</groupId>

            <artifactId>spring-kafka</artifactId>

        </dependency>

        <!-- 热部署 -->

        <dependency>

            <groupId>org.springframework</groupId>

            <artifactId>springloaded</artifactId>

            <version>1.2.8.RELEASE</version>

            <scope>provided</scope>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-devtools</artifactId>

            <scope>provided</scope>

        </dependency>

    </dependencies>

    <build>

        <finalName>${project.artifactId}</finalName>

        <plugins>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-compiler-plugin</artifactId>

                <configuration>

                    <source>1.8</source>

                    <target>1.8</target>

                    <encoding>UTF-8</encoding>

                </configuration>

            </plugin>

            <plugin>

                <groupId>org.springframework.boot</groupId>

                <artifactId>spring-boot-maven-plugin</artifactId>

                <executions>

                    <execution>

                        <goals>

                            <goal>repackage</goal>

                        </goals>

                    </execution>

                </executions>

            </plugin>

        </plugins>

    </build>

</project>

3、KafkaStarter.java

package com.java;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

/**

 * 主启动类

 *

 * @author Logan

 * @version 1.0.0

 * @createDate 2019-05-07

 *

 */

@SpringBootApplication

public class KafkaStarter {

    public static void main(String[] args) {

        SpringApplication.run(KafkaStarter.class, args);

    }

}

4、MessageHandler.java

package com.java.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Component;

/**

 *

 * 任务处理器，监听kafka队列中的消息，消费并处理

 *

 * @author Logan

 * @version 1.0.0

 * @createDate 2019-05-07

 *

 */

@Component

public class MessageHandler {

    @KafkaListener(topics = { "test-topic" })

    public void handle(String message) {

        System.out.println("[ 处理器开始处理消息 ]" + System.currentTimeMillis());

        try {

            Thread.sleep(5000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        System.out.println(message);

        System.out.println("[ 处理器处理消息完成 ]" + System.currentTimeMillis());

    }

    @KafkaListener(topics = { "test-topic" })

    public void handle(ConsumerRecord<String, String> record) {

        System.out.println("[ 处理器开始处理消息 ]" + System.currentTimeMillis());

        try {

            Thread.sleep(5000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        System.out.println(record);

        System.out.println("[ 处理器处理消息完成 ]" + System.currentTimeMillis());

    }

}

5、SendMessageController.java

package com.java.controller;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

/**

 * 发送消息类

 *

 * @author Logan

 * @version 1.0.0

 * @createDate 2019-05-07

 *

 */

@RestController

public class SendMessageController {

    @Autowired

    private KafkaTemplate<String, String> kafkaTemplate;

    private String topic = "test-topic";

    @GetMapping("/send")

    public String send(String params) {

        System.out.println("[ 收到请求 ]");

        kafkaTemplate.send(topic, params);

        System.out.println("[ 返回响应 ]");

        return "您的任务已提交";

    }

}

6、application.properties

#生产者配置，参见org.springframework.boot.autoconfigure.kafka.KafkaProperties.Producer

spring.kafka.producer.bootstrapServers=s1:9092,s2:9092,s3:9092

#消费者配置，参见org.springframework.boot.autoconfigure.kafka.KafkaProperties.Consumer

spring.kafka.consumer.bootstrapServers=s1:9092,s2:9092,s3:9092

spring.kafka.consumer.groupId=kafka-test

spring.kafka.consumer.autoOffsetReset=latest

spring.kafka.consumer.enableAutoCommit=true

7、运行KafkaStarter.java 启动

浏览器输入：http://127.0.0.1:8080/send?params=Good

可以向主题中发送消息Good，（params参数就是发送的内容）

程序中的消费者会监听到消息并开始处理

在Kafka消费者控制台可以监听到同样的消息

同样，在Kafka生产者控制台发送消息到主题 test-topic ，

程序中的消费者也会监听到消息并开始处理。

结论：测试生产者和消费者功能一切正常！

Spring boot 集成Kafka

.