zookeeper和kafka的SASL认证以及生产实践-阿里云开发者社区

一、什么是zookeeper？

ZooKeeper是一个集中的服务，用于维护配置信息、命名、提供分布式同步以及提供组服务。所有这些类型的服务都以某种形式被分布式应用程序使用。每次它们被实现时，都有大量的工作需要去修复不可避免的bug和竞争条件。由于实现这类服务的困难，应用程序最初通常会略过它们，这使得它们在出现变化时变得脆弱，难以管理。即使做得正确，这些服务的不同实现在部署应用程序时也会导致管理复杂性。

总之：zookeeper是一种高可靠的分布式协调组件，主要用来解决分布式一致性和分布式锁、元数据/配置信息管理的问题。

二、zookeeper的使用场景

分布式协调：这个其实是zk很经典的一个用法，简单来说，就好比，你A系统发送个请求到mq，然后B消息消费之后处理了。那A系统如何知道B系统的处理结果？用zk就可以实现分布式系统之间的协调工作。A系统发送请求之后可以在zk上对某个节点的值注册个监听器，一旦B系统处理完了就修改zk那个节点的值，A立马就可以收到通知，完美解决。
分布式锁：对某一个数据连续发出两个修改操作，两台机器同时收到了请求，但是只能一台机器先执行另外一个机器再执行。那么此时就可以使用zk分布式锁，一个机器接收到了请求之后先获取zk上的一把分布式锁，就是可以去创建一个znode，接着执行操作；然后另外一个机器也尝试去创建那个znode，结果发现自己创建不了，因为被别人创建了,那只能等着，等第一个机器执行完了自己再执行。
元数据/配置信息管理：zk可以用作很多系统的配置信息的管理，比如kafka、storm、Doubbo等等很多分布式系统都会选用zk来做一些元数据、配置信息的管理

三、zookeeper节点有哪些特性，什么时候使用什么特性的节点？

在 ZooKeeper 中，节点类型可以分为持久节点（PERSISTENT ）、临时节点（EPHEMERAL），以及有序节点（SEQUENTIAL ），具体在节点创建过程中，一般是组合使用。

持久节点：是指在节点创建后，就一直存在，直到有删除操作来主动清除这个节点——不会因为创建该节点的客户端会话失效而消失。
持久顺序节点：这类节点的基本特性和上面的节点类型是一致的。额外的特性是，在ZK中，每个父节点会为他的第一级子节点维护一份时序，会记录每个子节点创建的先后顺序。基于这个特性，在创建子节点的时候，可以设置这个属性，那么在创建节点过程中，ZK会自动为给定节点名加上一个数字后缀，作为新的节点名。这个数字后缀的范围是整型的最大值。
临时节点：与持久节点不同的是，临时节点的生命周期和客户端会话绑定。也就是说，如果客户端会话失效，那么这个节点就会自动被清除掉。注意，这里提到的是会话失效，而非连接断开。另外，在临时节点下面不能创建子节点。
事件监听：在读取数据时，我们可以同时对节点设置事件监听，当节点数据或结构变化时，zookeeper会通知客户端。当前zookeeper有如下四种事件：1)节点创建;2)节点删除;3)节点数据修改;4)子节点变更。

四、使用zookeeper实现服务注册中心，原理是什么？用到了zk的哪些特性？

zookeeper可以充当一个服务注册表（Service Registry），让多个服务提供者形成一个集群，让服务消费者通过服务注册表获取具体的服务访问地址（ip+端口）去访问具体的服务提供者。具体来说，zookeeper就是个分布式文件系统，每当一个服务提供者部署后都要将自己的服务注册到zookeeper的某一路径上,每当一个服务启动后会将自己的服务名称、IP等信息注册到zookeeper的ZNode上。
在这里插入图片描述

服务调用者第一次请求过来的时候会返回集群的IP地址的集合并注册一个服务的监听，然后在前端通过负载均衡算法(随机，轮询，加权轮询，最小连接数，hash)将对应的请求发到对应的设备上，而zookeeper和服务提供者之间会存在一个心跳机制，定时发送一个数据包来检查该服务器的状态，当接收不到数据包时，则会将该设备注册的相对应的服务进行删除，然后会push到服务调用者将对应的ip进行删除防止后续请求转发到已经宕掉的服务器上。

使用了数据节点持久化，事件监听的特性，分布式一致性特性

五、zookeeper集群搭建

多年之前我写了一篇简易的zookeeper入门教程，如果是搭建伪集群的话可以参考以上这篇文章，地址：[https://zhupei.blog.csdn.net/article/details/51517460
](https://zhupei.blog.csdn.net/article/details/51517460 "https://zhupei.blog.csdn.net/article/details/51517460")

下面是搭建生产环境的步骤：

准备3台服务器，例如node1、node2、node3，首先进入node1中

1、下载zk3.5版本的，http://www.apache.org/dyn/closer.cgi/zookeeper/
2、解压到相应目录,并在目录下新建dataDir文件夹

3、在conf目录在做配置：先拷贝一份 zoo_sample.cfg为zoo.cfg，然后编辑zoo.cfg：


    tickTime=2000
    initLimit=10
    syncLimit=5
    dataDir=/mnt/tools/zookeeper/apache-zookeeper-3.5.5/dataDir/
    clientPort=2181
    server.1=0.0.0.0:2888:3888
    server.2=node2:2888:3888
    server.3=node3:2888:3888

这里注意，node2、node3替换为真实服务器的ip地址，其中在node1中，这个ip地址要设置为0.0.0.0，不然肯会出现Cannot open channel to 1 at election address /xx.xx.xx.xx:3888的错误，特别是在真实的云服务器上面很容易出现这个问题。

4、参照前面3步，分别在node2,node3机器上面进行配置，在node2中cong文件就修改为：
```
    server.1=node1:2888:3888
    server.2=0.0.0.0:2888:3888
    server.3=node3:2888:3888
```

5、在node3中cong文件就修改为：

    server.1=node1:2888:3888
    server.2=node2:2888:3888
    server.3=0.0.0.0:2888:3888

6、启动，分别在3个节点执行进行启动
```
bin/zkServer.sh  start
```
7、查看状态，可以看到3个节点随机有是1个master和2个follow
```
  `bin/zkServer.sh  status`
```

六、kafka集群搭建

因为本文主要讲zookeeper，所以对于kafka的搭建的就简易的说一下。这里提到kafka是因为下面要讲到到zookeeper和kafka的SASL机制认证。
安装kafka集群之前，确保zookeeper服务已经正常运行。准备3台服务器，例如node1、node2、node3，首先进入node1中

1、下载kafka,地址 http://kafka.apache.org/downloads
2、解压到相应目录，并在安装目录下新建logs目录
3、编辑config中的server.properties文件，注意这里的broker.id，部署在不同机器上面的id需要不同，例如在node1中部署，这里就填写1，在node2机器上面就填写2，依次类推。

broker.id=1
listeners=SASL_PLAINTEXT://0.0.0.0:9092
log.dirs=/mnt/tools/kafka/kafka_2.12/logs
num.partitions=3
zookeeper.connect=node1:2181,node2:2181,node3:2181

4、以上设置都完毕，保存配置并退出，然后将kafka目录发送至其他主机
5、在各个节点上启动

bin/kafka-server-start.sh -daemon config/server.properties

6、通过jps可以看到有kafka的进程以及查看日志都正常的话就表示启动成功了

七、zookeeper和kafka的安全认证机制SASL

zookeeper在生产环境中，如果不是只在内网开放的话，就需要设置安全认证，可以选择SASL的安全认证。以下是和kafka的联合配置，如果不需要kafka可以去掉kafka相关的权限即可，以下基于zk3.5.5和kafka2.12进行操作。
下面就是详细的部署步骤：

7.1 zookeeper的安全认证配置

zookeeper所有节点都是对等的，只是各个节点角色可能不相同。以下步骤所有的节点配置相同。

1、导入kafka的相关jar
从kafka/lib目录下复制以下几个jar包到zookeeper的lib目录下：

kafka-clients-2.3.0.jar
lz4-java-1.6.0.jar
slf4j-api-1.7.25.jar
slf4j-log4j12-1.7.25.jar
snappy-java-1.1.7.3.jar

2、zoo.cfg文件配置
添加如下配置：


authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

3、编写JAAS文件，zk_server_jaas.conf，放置在conf目录下
这个文件定义需要链接到Zookeeper服务器的用户名和密码。JAAS配置节默认为Server：

Server {
org.apache.kafka.common.security.plain.PlainLoginModule required 
    username="admin" 
    password="admin-2019" 
    user_kafka="kafka-2019" 
    user_producer="prod-2019";
};

这个文件中定义了两个用户，一个是kafka，一个是producer，这些用user_配置出来的用户都可以提供给生产者程序和消费者程序认证使用。还有两个属性，username和password，其中username是配置Zookeeper节点之间内部认证的用户名，password是对应的密码。

4、修改zkEnv.sh
在zkEnv.sh添加以下内容，路径按你直接的实际路径来填写：

export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/mnt/tools/zookeeper/apache-zookeeper-3.5.5/conf/zk_server_jaas.conf "

5、在各个节点分别执行bin/zkServer.sh start启动zk。如果启动异常查看日志排查问题。

7.2 kafka的安全认证配置

zookeeper启动之后，就配置kafka,下面步骤的配置在所有节点上都相同。

1、在kafka的config目录下，新建kafka_server_jaas.conf文件，内容如下：

KafkaServer {
      org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-2019"
    user_admin="admin-2019"
    user_producer="prod-2019"
    user_consumer="cons-2019";
};

Client {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafka-2019";
};

KafkaServer配置的是kafka的账号和密码，Client配置节主要配置了broker到Zookeeper的链接用户名密码，这里要和前面zookeeper配置中的zk_server_jaas.conf中user_kafka的账号和密码相同。

2、配置server.properties，同样的在config目录下

listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://node1: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

这里注意listeners配置项，将主机名部分（本例主机名是node1）替换成当前节点的主机名。其他在各个节点的配置一致。注意，allow.everyone.if.no.acl.found这个配置项默认是false，若不配置成true，后续生产者、消费者无法正常使用Kafka。

3、在server启动脚本JVM参数,在bin目录下的kafka-server-start.sh中，将

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

这一行修改为

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/mnt/tools/kafka/kafka_2.12/config/kafka_server_jaas.conf"

4、配置其他节点
配置剩余的kafka broker节点，注意server.properties的listeners配置项

5.启动各个节点的kafka服务端，在bin目录下执行

./kafka-server-start.sh ../config/server.properties

八、springboot中使用kafka

在springboot项目中，注意：如果在springboot1.x的版本中报错，请升级为speingboot2.x的版本，本例使用的是2.1.1.RELEASE

1、新建sprinboot工程，添加pom依赖，设置parent为2.1.1.RELEASE,这样就不需要在dependency中添加版本号了

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

2、在resource下新建文件kafka_client_jaas.conf，内容如下：

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

这个用户名和密码要和前面kafka中使用的账号密码相同，这样才能有访问权限

3、在application.yml文件配置,注意这里将node1,2,3替换为你真实的服务器地址

spring:
  kafka:
    listener:
      batch-listener: true                                #是否开启批量消费，true表示批量消费
      concurrency: 10                                      #设置消费的线程数
      poll-timeout: 1500                                  #自动提交设置，如果消息队列中没有消息，等待timeout毫秒后，调用poll()方法。如果队列中有消息，立即消费消息，每次消费的消息的多少可以通过max.poll.records配置。
    template:
      default-topic: probe2
    producer:
      bootstrap-servers: node1:9092,node2,node3:9092
      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: 
      bootstrap-servers: node1:9092,node2:9092,node3:9092
      group-id: group-2
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties: 
        sasl.mechanism: PLAIN
        security.protocol: SASL_PLAINTEXT
      auto:
        offset: 
            reset: latest
      enable:
        auto: 
          commit: true

在这里开启了消费者10个线程，这里可以根据你实际业务来调整消费者的数量

4、程序启动类：


@SpringBootApplication
public class ProbeApplication {
    
    //初始化系统属性
    static {
        ClassLoader loader = Thread.currentThread().getContextClassLoader();
        System.setProperty("java.security.auth.login.config", 
                loader.getResource("").getPath()+File.separator+"kafka_client_jaas.conf");
    }

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

5、生产者：

@Component
public class Sender {
    private static final Logger LOG = LoggerFactory.getLogger(Sender.class);
    @Autowired
    private KafkaTemplate<String, String> template;
    
    public void send(String message) {
        ListenableFuture<SendResult<String, String>> future = this.template.sendDefault(message);
        future.addCallback(success -> LOG.info("KafkaMessageProducer 发送消息成功！消息内容是:"+message),
                fail -> LOG.error("KafkaMessageProducer 发送消息失败！消息内容是:"+message));
    }
}

6、消费者，将生产者发过来的消息进行处理

@Component
public class Receiver {
    
    @KafkaListener(topics = "probe2")
    public void receiveMessage(ConsumerRecord<String, String> record) {
            System.out.println("【*** 消费者开始接收消息 ***】key = " + record.key() + "、value = " + record.value());
           //TODO，在这里进行自己的业务操作，例如入库
    }
}

7、controller

@RestController
public class KafkaController {
    @Autowired
    private Sender sender;

    @PostMapping("/send/{msg}")
    public String send(@PathVariable("msg") String msg) {
        sender.send(msg);
        return msg;
    }
}

8、启动springboot工程，然后访问相应的地址即可得到想要的结果

如果生产者的速度大于消费者的速度，可以适当调整生产者和消费者的数量来处理，同时不要在消费者进行太过于耗时的操作。

九、总结

本文主要分享了zookeeper的应用场景和节点特性、注册原理、zookeeper集群搭建和kafka集群搭建、zookeeper和kafka的SASL认证机制、在springboot中实操基于SASL认证的kafka。

zookeeper和kafka的SASL认证以及生产实践

一、什么是zookeeper？

二、zookeeper的使用场景

三、zookeeper节点有哪些特性，什么时候使用什么特性的节点？

四、使用zookeeper实现服务注册中心，原理是什么？用到了zk的哪些特性？

五、zookeeper集群搭建

六、kafka集群搭建

七、zookeeper和kafka的安全认证机制SASL

7.1 zookeeper的安全认证配置

7.2 kafka的安全认证配置

八、springboot中使用kafka

九、总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

zookeeper和kafka的SASL认证以及生产实践

一、什么是zookeeper？

二、zookeeper的使用场景

三、zookeeper节点有哪些特性，什么时候使用什么特性的节点？

四、使用zookeeper实现服务注册中心，原理是什么？用到了zk的哪些特性？

五、zookeeper集群搭建

六、kafka集群搭建

七、zookeeper和kafka的安全认证机制SASL

7.1 zookeeper的安全认证配置

7.2 kafka的安全认证配置

八、springboot中使用kafka

九、总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景