【KAFKA】Kafka集群搭建(安全认证SASL/PLAIN),并和springBoot整合

我就是我 2022-04-24 08:16 412阅读 0赞

# Kafka集群搭建(安全认证SASL/PLAIN),并和springBoot整合 #

搭建三台机器的kafka集群。 用外置的zookeeper做配置中心管理。 为了不让配置中心也产生单点，所以我们需要先搭建一个三个节点的配置中心集群；

## 1 环境： ##

OS版本：        Centos7.4  
kafka版本：    kafka\_2.11-2.2.0  
ZK版本:         zookeeper-3.4.14  
JDK版本:         jdk\_1.8

服务器列表（此处设置了域名）：  
192.168.0.85    gmg\_server02  
192.168.0.86    gmg\_server03  
192.168.0.90    gmg\_server04

安装目录：  
/home/programs/zookeeper  
/home/programs/kafka  
/home/programs/jdk

## 2 服务端设置 ##

### 2.1 安装和配置zookeeper ###

请参照：https://blog.csdn.net/shgh\_2004/article/details/89282066

### 2.2 安装和配置kafka ###

**2.2.1 解压**  
tar -zxvf kafka\_2.11-2.2.0.tgz -C /home/programs/kafka/  
对于单机系统，解压好了即可直接运行zookeeper和kafka服务。

**2.2.2 修改配置文件**  
a) 配置kafka（server.properties）

vi conf/server.properties
    
    # 设置broker的唯一标识（集群唯一随便设置）。
    broker.id=2 
    
    # 设置带SASL安全认证的服务监听（域名或者IP修改为机器的正式IP或者域名就好 其他照抄）
    listeners=SASL_PLAINTEXT://gmg_server02:9092
    security.inter.broker.protocol=SASL_PLAINTEXT
    sasl.enabled.mechanisms=PLAIN
    sasl.mechanism.inter.broker.protocol=PLAIN
    authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
    allow.everyone.if.no.acl.found=true
    
    # 设置log的目录 （这个在清理掉topic的时候会有用）
    log.dirs=/home/programs/kafka/data/logs
    
    # 连接的zookeeper 集群地址列表
    zookeeper.connect=gmg_server01:2181,gmg_server02:2181,gmg_server03:2181

**b) 创建JAAS配置文件（kafka的认证账户信息）**

定义链接Kafka Broker时所需要的用户名密码及broker各个节点之间相互通信的用户名密码，这部分配置定义在KafkaServer节，文件如下：

touch config/kafka_server_jaas.conf
    KafkaServer {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin-shiguanghui"
        user_admin="admin-shiguanghui";
    };

**c) 配置KAFKA\_OPTS环境变量**

为KAFKA_OPTS设置启动参数。
    export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G"
    增加如下：
    export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G -Djava.security.auth.login.config=/home/programs/kafka/config/kafka_server_jaas.conf"

### 2.3 启动服务 ###

首先启动zookeeper机器。 然后轮流启动kafka集群。  
/home/programs/kafka/bin/kafka-server-start.sh -daemon /home/programs/kafka/config/server.properties

##    3 客户端设置 ##

这里的没有做kafka自带客户端的设置，主要演示一下应用程序客户端的设置。 我们是使用springboot的应用客户端。

### 3.1 项目中引入依赖： ###

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

**3.2 在resources下面新建文件：kafka\_client\_jaas.conf**

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

**3.3 application.yml的配置**

spring:
      kafka:
          bootstrap-servers: gmg_server01:9092,gmg_server02:9092,gmg_server03:9092
          topic-name: test-topic
        consumer:
            group-id: prd_group
          # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
          # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
          # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
          auto-offset-reset: latest
          max-poll-records: 100
          concurrency: 10
          # key/value的反序列化
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          properties:
            sasl.mechanism: PLAIN
            security.protocol: SASL_PLAINTEXT
        consumer-extra:
          # 是否批量处理
          batch-listener: true
        producer:
          # key/value的序列化
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          properties:
            sasl.mechanism: PLAIN
            security.protocol: SASL_PLAINTEXT
          # 批量抓取
          batch-size: 4096
          # 缓存容量
          buffer-memory: 40960

**3.5 程序启动类：**

@SpringBootApplication
    public class APIApplication extends SpringBootServletInitializer {
        static {
                // 这里可能会有坑。因为这种方式springboot无法读取jar包中配置文件。需要自行处理
                String config = ResourceUtils.getFile("classpath:kafka_client_jaas.conf").getAbsolutePath();
            System.setProperty("java.security.auth.login.config", config);
        }
    
        public static void main(String[] args) {
            SpringApplication.run(APIApplication.class, args);
        }
    }

**3.6 消息生产者**

@Component
    public class Sender {
        @Autowired
        private KafkaTemplate<String, String> template;
    
        public void send(String msg) {
    
            this.template.sendDefault("test-msg", msg);
            System.out.println("发送消息：" + msg);
        }
    }

**3.6 消息消费者**

@Component
    public class Receiver {
        @KafkaListener(topics = {${spring.kafka.consumer.topic-name}}, groupId = "${spring.kafka.consumer.group-id}")
        public void receiveMessage(ConsumerRecord<String, String> record) {
            System.out.println("接收消息key = " + record.key() + "、value = " + record.value());
        }
    }

**3.7 controller **

@RestController
    public class KafkaController {
        @Autowired
        private Sender sender;
    
        @PostMapping("/{msg}")
        public String send(@PathVariable("msg") String msg) {
            sender.send(msg);
            return msg;
        }
    }