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

本文主要是介绍【KAFKA】Kafka集群搭建(安全认证SASL/PLAIN),并和springBoot整合，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

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 requiredusername="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 requiredusername="admin"password="admin-shiguanghui";
};

3.3 application.yml的配置

spring:kafka:bootstrap-servers: gmg_server01:9092,gmg_server02:9092,gmg_server03:9092topic-name: test-topicconsumer:group-id: prd_group# earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费# latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据# none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: latestmax-poll-records: 100concurrency: 10# key/value的反序列化key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproperties:sasl.mechanism: PLAINsecurity.protocol: SASL_PLAINTEXTconsumer-extra:# 是否批量处理batch-listener: trueproducer:# key/value的序列化key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerproperties:sasl.mechanism: PLAINsecurity.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 {@Autowiredprivate 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 {@Autowiredprivate Sender sender;@PostMapping("/{msg}")public String send(@PathVariable("msg") String msg) {sender.send(msg);return msg;}
}

 

这篇关于【KAFKA】Kafka集群搭建(安全认证SASL/PLAIN),并和springBoot整合的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---