本文主要介绍基于SCRAM进行身份验证，使用Kafka ACL进行授权，SSL进行加密以及使用camel-Kafka连接Kafka群集以使用camel路由生产和消费消息的过程。
加密与认证
AMQ Streams支持加密和身份验证，这是侦听器配置的一部分。
侦听器配置
Kafka代理中的加密和身份验证是针对每个侦听器配置的。
Kafka代理中的每个侦听器都配置有自己的安全协议。配置属性"listener.security.protocal"定义哪个侦听器使用哪种安全协议。然后将每个侦听器名称映射到其安全协议中。
支持的安全协议有：
- 纯文本：侦听器，无需任何加密或身份验证。
- SSL协议：使用TLS加密的侦听器，以及使用TLS客户端证书的身份验证（可选）。
- SASL_PLAINTEXT：侦听器不加密，但具有基于SASL的身份验证。
- SASL_SSL：具有基于TLS的加密和基于SASL的身份验证的侦听器。
以下为指定SASL_SSL协议侦听器配置：
代码语言：javascript复制listeners=SASL_SSL://localhost:9092
advertised.listeners=SASL_SSL://localhost:9092
security.inter.broker.protocol=SASL_SSLTLS加密
为了使用TLS加密和服务器身份验证，必须提供包含私钥和公钥的密钥库。通常使用Java密钥存储（JKS）格式的文件来完成此操作。在"ssl.keystore.location"属性中设置此文件的路径。"ssl.keystore.password"为集群中的所有Kafka代理生成TLS证书。证书应在其通用名称或主题备用名称中具有其公告的地址和引导地址。
编辑所有Kafka集群节点上的/opt/kafka/config/server.properties 配置文件，以进行以下操作：
更改"listener.security.protocol.map"字段以为要使用TLS加密的侦听器指定SSL协议。
将"ssl.keystore.location"选项设置为带有代理证书的JKS密钥库的路径。
将"ssl.keystore.password"选项设置为用于保护密钥库的密码。
SASL认证
使用Java身份验证和授权服务（JAAS）配置SASL身份验证。JAAS还用于验证Kafka和ZooKeeper之间的连接。
JAAS使用其自己的配置文件。该文件位于/opt/kafka/config/jaas.conf，通过普通的未加密连接以及通过TLS连接都支持SASL身份验证。可以分别为每个侦听器启用SASL。要启用它，listener.security.protocol.map中的安全协议必须是SASL_PLAINTEXT或SASL_SSL。
Kafka中的SASL身份验证支持几种不同的机制：
- 普通
根据用户名和密码实施身份验证。用户名和密码以Kafka配置存储在本地。
- SCRAM-SHA-256和SCRAM-SHA-512
使用Salted Challenge Response身份验证机制（SCRAM）实现身份验证。SCRAM凭证集中存储在ZooKeeper中。SCRAM可以用于ZooKeeper群集节点在专用网络中隔离运行的情况。
- GSSAPI
针对Kerberos服务器实施身份验证
通过JAAS配置文件配置SASL机制。Kafka使用名为Kafka服务器的JAAS上下文。在JAAS中配置它们之后，必须在Kafka配置中启用SASL机制。这可以使用sasl.enabled.mechanisms属性完成
- SASL SCRAM
Kafka中的SCRAM身份验证包含两种机制：SCRAM-SHA-256和SCRAM-SHA-512。这些机制仅在使用的哈希算法上有所不同-SHA-256与更强的SHA-512。要启用SCRAM身份验证，JAAS配置文件必须包含以下配置：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/kafka_server_jass.conf
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin123";
};并在server.properties文件中启用SASL身份验证:
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/server.properties
# With SASL & SSL encryption
scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin@2016.08.19.com";在$kafka-home/config目录下创建ssl-user-config.properties文件，如下所示：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]% vi ${kafka_home}/config/ssl-user-config.properties
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
ssl.truststore.location=$kafka_home/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.comSCRAM机制的用户凭据存储在ZooKeeper中。 可借助kafka-configs.sh工具进行管理。
代码语言：javascript复制 [administrator@JavaLangOutOfMemory ~ ]%/bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='admin@2016.08.19.com']' --entity-type users --entity-name admin
Warning: --zookeeper is deprecated and will be removed in a future version of Kafka.
Use --bootstrap-server instead to specify a broker to connect to.
Completed updating config for entity: user-principal 'admin'.
./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='luga@2016.08.19.com']' --entity-type users --entity-name luga
Warning: --zookeeper is deprecated and will be removed in a future version of Kafka.
Use --bootstrap-server instead to specify a broker to connect to.
Completed updating config for entity: user-principal 'luga'.完整的$ {kafka-home} /config/server.properties文件如下所示:
代码语言：javascript复制# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# see kafka.server.KafkaConfig for additional details and defaults
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=SASL_SSL://localhost:9092
# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=SASL_SSL://localhost:9092
security.inter.broker.protocol=SASL_SSL
ssl.endpoint.identification.algorithm=
ssl.client.auth=required
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512
# Broker security settings
46
ssl.truststore.location=/${kafka_home}/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com
ssl.keystore.location=/${kafka_home}/config/keystore/kafka.keystore.jks
ssl.keystore.password=luga@2016.08.19.com
ssl.key.password=luga@2016.08.19.com
# ACLs
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
# #zookeeper SASL
zookeeper.set.acl=false
# # With SASL & SSL encryption
scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin123";
# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3
# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1
############################# Internal Topic Settings #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Flush Policy #############################
# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
# 1. Durability: Unflushed data may be lost if you are not using replication.
# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.
# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=18000
############################# Group Coordinator Settings #############################
# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0用JAAS启动Kafka，命令如下所示：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]%export KAFKA_OPTS=-Djava.security.auth.login.config=/${kafka_home}/config/kafka_server_jaas.conf
[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-server-start.sh /${kafka_home}/config/server.properties创建主题Demo-Topic:
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --operation Create --operation Describe --topic demo-topic
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW) 同理，我们给生产者和消费者授予相关权限：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --producer --topic demo-topic
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:luga --consumer --topic demo-topic --group demo-consumer-group
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=READ, permissionType=ALLOW)
(principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
Adding ACLs for resource `ResourcePattern(resourceType=GROUP, name=demo-consumer-group, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=READ, permissionType=ALLOW)
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo-topic, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=CREATE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=WRITE, permissionType=ALLOW)
(principal=User:luga, host=*, operation=READ, permissionType=ALLOW)
Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=demo-consumer-group, patternType=LITERAL)`:
(principal=User:luga, host=*, operation=READ, permissionType=ALLOW)开始生产消息：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic demo-topic --producer.config config/ssl-producer.properties
>Hi this is docker
>生产者配置ssl-producer.properties：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]% vi /${kafka_home}/config/ssl-producer.properties
bootstrap.servers=localhost:9092
compression.type=none
### SECURITY ######
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
ssl.truststore.location=/{kafka_home}/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com消费者配置ssl-consumer.properties：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]% vi /${kafka_home}/config/ssl-consumer.properties
bootstrap.servers=localhost:9092
# consumer group id
group.id=test-consumer-group
# What to do when there is no initial offset in Kafka or if the current
# offset does not exist any more on the server: latest, earliest, none
#auto.offset.reset=
### SECURITY ######
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";
ssl.truststore.location=/${kafka_home}/config/truststore/kafka.truststore.jks
ssl.truststore.password=luga@2016.08.19.com开始消费消息：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]%./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo-topic --from-beginning --consumer.config config/ssl-consumer.properties现在基于came路由启动Spring Boot应用程序进行消息的生产与消费：
代码语言：javascript复制public class KafkaRouteBuilder extends RouteBuilder {
@Override
public void configure() throws Exception {
//producer
from("timer://foo?period={{period}}")
.setBody(constant("Hi This is kafka example"))
.to("kafka:{{kafka.topic}}?brokers={{kafka.bootstrap.url}}"
"&keySerializerClass=org.apache.kafka.common.serialization.StringSerializer"
"&serializerClass=org.apache.kafka.common.serialization.StringSerializer"
"&securityProtocol={{security.protocol}}&saslJaasConfig={{sasl.jaas.config}}"
"&saslMechanism={{sasl.mechanism}}&sslTruststoreLocation={{ssl.truststore.location}}"
"&sslTruststorePassword={{ssl.truststore.password}}&sslTruststoreType={{ssl.truststore.type}}")
.log("${body}");
//consumer
from("kafka:{{consumer.topic}}?brokers={{kafka.bootstrap.url}}&maxPollRecords={{consumer.max.poll.records}}"
"&keySerializerClass=org.apache.kafka.common.serialization.StringSerializer"
"&serializerClass=org.apache.kafka.common.serialization.StringSerializer"
"&groupId={{consumer.group}}&securityProtocol={{security.protocol}}&saslJaasConfig={{sasl.jaas.config}}"
"&saslMechanism={{sasl.mechanism}}&sslTruststoreLocation={{ssl.truststore.location}}"
"&sslTruststorePassword={{ssl.truststore.password}}&sslTruststoreType={{ssl.truststore.type}}"
"&autoOffsetReset={{consumer.auto.offset.reset}}&autoCommitEnable={{consumer.auto.commit.enable}}")
.log("${body}");
}
application.yml文件配置如下：
代码语言：javascript复制kafka.topic=demo-topic
kafka.bootstrap.url=localhost:9092
period=10000&repeatCount=5
kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
kafka.value.serializer=org.apache.kafka.common.serialization.StringSerializer
consumer.topic=demo-topic
consumer.group=demo-consumer-group
consumer.max.poll.records=1
consumer.threads=10
consumer.consumersCount=1
consumer.auto.offset.reset=earliest
consumer.auto.commit.enable=true
consumer.receive.buffer.bytes=-1
security.protocol = SASL_SSL
ssl.truststore.location =/luga/git/{...}/src/main/truststore/kafka.truststore.jks
ssl.truststore.password =luga@2016.08.19.com
ssl.truststore.type = JKS
sasl.mechanism = SCRAM-SHA-512
#sasl.kerberos.service.name=
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="luga" password="luga@2016.08.19.com";测试运行：
代码语言：javascript复制[administrator@JavaLangOutOfMemory ~ ]% mvn spring-boot:run
[INFO] Scanning for projects...
[INFO]
[INFO] ---------< org.apache.camel.example:camel-example-kafka-sasl >----------
[INFO] Building Camel :: Example :: Kafka :: sasl 1.0
[INFO] --------------------------------[ jar ]---------------------------------
。。。至此，简单的场景运行起来。
