早餐一个白鸡蛋吃得俺打嗝  -_-|||

准备工作

1、启动需要Kerberos认证的kafka。

注：具体配置方式可详见https://blog.csdn.net/justry_deng/article/details/88386114。

2、在本地将C:\Windows\System32\drivers\etc\hosts中指定kafka所在broker的ip地址以及其对应的主机名。

提示：本人的Java代码，测试时是写在Windows上的，而kafka是部署在虚拟机Linux上的。

3、配置客户端的kafka_client_jaas.conf认证文件。

创建一个kafka_client_jaas.conf文件，并令其内容为：

KafkaClient {  
        org.apache.kafka.common.security.plain.PlainLoginModule required  
        username="bob"  
        password="bob-pwd";  
};

注：本人放置客户端jaas认证文件的位置在：C:/Users/JustryDeng/Desktop/kerberos/kafka_client_jaas.conf。

注：这里指定了当前用户为bob。且此处的用户不能乱配置，只能使用(服务端)配置Kerberos认证时在
        kafka_server_jaas.conf认证文件中配置了的用户之一。

SpringBoot整合并简单使用Kerberos认证的Kafka

第一步：在application.properties系统配置文件中配置Kafka相关参数。

# -> -> -> -> -> -> -> ->  -> ->  -> -> kafka 基础配置
# 指定kafka 代理地址，可以多个，以逗号分隔
spring.kafka.bootstrap-servers=kafka-single:9095
# -> -> -> -> -> -> -> ->  -> ->  -> -> kafka 消息生产者
# 重试次数
spring.kafka.producer.retries=0
# -> -> -> -> -> -> -> ->  -> ->  -> -> kafka 消息消费者
# 偏移量策略设置
# earliest(当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费)
# latest(当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据)
# none(topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常)
spring.kafka.consumer.auto-offset-reset=earliest
# 允许自动提交消费者偏移量
spring.kafka.consumer.enable-auto-commit=true

# -> -> -> -> -> -> -> ->  -> ->  -> -> Kerberos
# Kerberos基础配合配置
spring.kafka.jaas.enabled=true
spring.kafka.properties.sasl.mechanism = PLAIN
spring.kafka.properties.security.protocol = SASL_PLAINTEXT

第二步：在pom.xml中引入相关依赖。

<!-- 需要引入与所安装的kafka对应版本的依赖 -->
<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.2.2.RELEASE</version>
</dependency>

注：引入什么版本的依赖需要看安装的是什么版本的Kafka。

更多对应关系可详见http://spring.io/projects/spring-kafka。

注：本人安装的Kafka版本是2.1.0的，所以会这里引入的依赖时2.2.x的spring-kafka。

这里给出完整的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>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.aspire</groupId>
    <artifactId>kafka</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka</name>
    <description>java使用kafka，简单示例</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

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

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

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <!-- 需要引入与所安装的kafka对应版本的依赖 -->
        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.2.RELEASE</version>
        </dependency>
    </dependencies>

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

</project>

第三步：编写消费者生产者等相关类

项目结构说明：

注：从上往下分别是消费者、生产者、启动项目时会进行消息发送的Runner类、启动类、配置文件、pom.xml文件。

生产者(示例)：

package com.single;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * 生产者
 *
 * @author JustryDeng
 * @date 2019/2/15 9:44
 */
@Component
public class MyProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public MyProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 发送消息
     * 注:kafkaTemplate.send(param...)是一个异步的方法，其发送结果可以通过Future的实现来获得。
     *
     * @author JustryDeng
     * @date 2019/2/15 12:01
     */
    public void sendMessage(String message){
        String topic = "topicOne";
        // 将消息发送到指定的主题下；这里可以是一个不存在的主题，会自动创建(注:自动创建的主题默认的分区数量是1个)
        kafkaTemplate.send(topic, message);
    }
}

注：KafkaTemplate提供了多个发送消息的方法，上例所示只是其一。

消费者：

package com.single;

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

/**
 * 消费者
 *
 * @author JustryDeng
 * @date 2019/2/15 9:44
 */
@Component
public class MyConsumer {

    /**
     * 注:@KafkaListener注解的属性很多，包括 匹配topic、错误处理、分组等等，
     *    实际使用时根据项目情况灵活运用即可。
     *
     * @author JustryDeng
     * @date 2019/2/15 12:30
     */
    @KafkaListener(topics = {"topicOne", "topicB"}, groupId = "group-01")
    public void consumerOne(String message) {
        System.out.println("consumerOne消费了消息 -> " + message);
    }

    /**
     * 注:@KafkaListener注解的属性很多，包括 匹配topic、错误处理、分组等等，
     *    实际使用时根据项目情况灵活运用即可。
     *
     * @author JustryDeng
     * @date 2019/2/15 12:30
     */
    @KafkaListener(topics = {"topicOne", "topicB"}, groupId = "group-01")
    public void consumerTwo(String message) {
        System.out.println("consumerTwo消费了消息 -> " + message);
    }

    /**
     * 注:@KafkaListener注解的属性很多，包括 匹配topic、错误处理、分组等等，
     *    实际使用时根据项目情况灵活运用即可。
     *
     * @author JustryDeng
     * @date 2019/2/15 12:30
     */
    @KafkaListener(topics = {"topicOne", "topicB"}, groupId = "group-02")
    public void consumerThree(String message) {
        System.out.println("consumerThree消费了消息 -> " + message);
    }
}

注：更多的时候，我们会使用ConsumerRecord<K, V>来接收消息，因为我们可以从该模型中获取到更多的信息，如：

ConsumerRecord<K, V>中封装的信息有：

启动类：

package com;

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

/**
 * 启动类
 *
 * @author JustryDeng
 * @date 2019/2/15 9:43
 */
@SpringBootApplication
public class KafkaApplication {

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

    /**
     * 系统环境属性 --- 设置
     *
     * 注:因为是系统参数，多出地方都要使用；所以直接写在启动类里面
     *
     * 注:设置系统环境属性 的 方式较多，这只是其中的一种
     *
     * @author JustryDeng
     * @date 2019/2/24 10:31
     */
    private static void systemPropertisConfig(){
        System.setProperty("java.security.auth.login.config",
                           "C:/Users/JustryDeng/Desktop/kerberos/kafka_client_jaas.conf");
    }

}

注：本人放置客户端jaas认证文件的位置在：C:/Users/JustryDeng/Desktop/kerberos/kafka_client_jaas.conf。

Runner测试类：

package com.single;

import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

/**
 * 测试消息生产者与消息消费者
 *
 *
 * @author JustryDeng
 * @date 2019/2/15 12:35
 */
@Component
public class MyRunner implements ApplicationRunner {

    private final MyProducer myProducer;

    public MyRunner(MyProducer myProducer) {
        this.myProducer = myProducer;
    }

    @Override
    public void run(ApplicationArguments args) {
        int count = 10;
        for (int i = 0; i < count; i++) {
            myProducer.sendMessage("我是消息, 我被发送出去啦！" + i);
        }
    }
}

测试一下

启动SpringBoot启动类，控制台输出：

 

注：【topicOne】主题是本人重新创建的只有一个分区的主题。所以同一个消费者组下，这批消息的消费只能
         有一个(consumerOne和consumerTwo属于同一个消费者组，consumerThree属于另一个消费者组。所以
         如上图所示，要么只有consumerOne和consumerThree，要么只有consumerOne和consumerThree)。

可见示例成功！

 

声明：本文为学习笔记，学习自51CTO，《Kafka消息中间件》，讲师李兴华。

^_^ 如有不当之处，欢迎指正

^_^ 学习视频：
              《Kafka消息中间件》，讲师李兴华

^_^ 测试代码托管链接：
              https://github.com/JustryDeng/CommonRepository

^_^ 本文已经被收录进《程序员成长笔记(四)》，笔者JustryDeng

Logo
CSDN学习社区

CSDN联合极客时间，共同打造面向开发者的精品内容学习社区，助力成长！

更多推荐

cover

用 OpenAI Assistants 做大模型应用开发

avatar CSDN学习社区
cover

1 小时解读鸿蒙 10 大热点问题

avatar CSDN学习社区
cover

1 小时解读鸿蒙 10 大热点问题

avatar CSDN学习社区