摘要：
对于消息组件，RabbitMQ根据原始设计具有最高的性能。因为RabbitMQ设计相对完整，但Kafka不同。卡夫卡最初设计的最大缺陷是没有安全认证。因此，为了考虑安全性，一些初始系统使用性能较差的ssl身份验证进行设计，即使用证书来处理身份验证操作。如果要对证书执行ssl处理，则必须使用Java中的keytool工具生成相应的证书信息，然后在访问之前将证书信息提交给客户端和服务器进行共同匹配。
对于消息组件，如果按照最初的设计来讲，性能最高得就是RabbitMQ，因为RabbitMQ设计比较完整，而Kafka不同，Kafka最初的设计最大的坑：没有安全认证，所以最初的一些系统设计的时候为了考虑到安全性，所以采用了性能较差的ssl认证方式来进行出来，也就是说使用证书的方式来处理认证操作。
如果要想进行证书的ssl处理，那么肯定需要通过java中的keytool工具生成相应的证书信息，而后将此证书信息内容交给客户端与服务端共同匹配后才可以访问。
一、生成ssl证书
1、 为了方便证书的保存处理，首先建立一个证书的保存目录：
mkdir -p /usr/ca/{root,server,client,trust}
2、 生成server.keystore.jks证书的孵化器：
keytool -keystore /usr/ca/server/server.keystore.jks -alias kafka-single -validity 365 -genkey -keypass mldnjava -dname "CN=kafka-single.com,OU=mldn,O=mldnjava,L=bj,S=bj,C=cn" -storepass mldnjava
在/usr/ca/server/目录下会生成server.keystore.jks文件
3、 为了保证整个证书的安全性，所以需要使用CA进行证书的签名保证，下面创建CA的认证证书：
openssl req -new -x509 -keyout /usr/ca/root/ca-key -out /usr/ca/root/ca-cert -days 365 -passout pass:mldnjava -subj "/C=cn/ST=bj/L=b/O=mldn/OU=mldnjava/CN=kafka-single.com"
4、 通过CA创建一个服务器端的信任证书，有了信任证书才可以进行证书有效性的检测：
keytool -keystore /usr/ca/trust/server.truststore.jks -alias CARoot -import -file /usr/ca/root/ca-cert -storepass mldnjava
提示时输入：y
5、 通过CA创建一个客户端的信任证书：
keytool -keystore /usr/ca/trust/client.truststore.jks -alias CARoot -import -file /usr/ca/root/ca-cert -storepass mldnjava
6、 进行服务器端证书签名处理：
##导出服务器证书"server.cert-file":
keytool -keystore /usr/ca/server/server.keystore.jks -alias kafka-single -certreq -file /usr/ca/server/server.cert-file -storepass mldnjava
##用ca给服务器证书进行签名处理：
openssl x509 -req -CA /usr/ca/root/ca-cert -CAkey /usr/ca/root/ca-key -in /usr/ca/server/server.cert-file -out /usr/ca/server/server.cert-signed -days 365 -CAcreateserial -passin pass:mldnjava
##将CA证书导入到服务器keystore
keytool -keystore /usr/ca/server/server.keystore.jks -alias CARoot -import -file /usr/ca/root/ca-cert -storepass mldnjava
##将已签名的服务器证书导入到服务器keystore
keytool -keystore /usr/ca/server/server.keystore.jks -alias kafka-single -import -file /usr/ca/server/server.cert-signed -storepass mldnjava
7、 如果你的现在kafka主机有很多（集群），那么所有的客户端服务器还需要设置客户端证书处理：
keytool -keystore /usr/ca/client/client.keystore.jks -alias kafka-single -validity 365 -genkey -keypass mldnjava -dname "CN=kafka-single.com,OU=mldn,O=mldnjava,L=bj,S=bj,C=cn" -storepass mldnjava
keytool -keystore /usr/ca/client/client.keystore.jks -alias kafka-single -certreq -file /usr/ca/client/client.cert-file -storepass mldnjava
openssl x509 -req -CA /usr/ca/root/ca-cert -CAkey /usr/ca/root/ca-key -in /usr/ca/client/client.cert-file -out /usr/ca/client/client.cert-signed -days 365 -CAcreateserial -passin pass:mldnjava
keytool -keystore /usr/ca/client/client.keystore.jks -alias CARoot -import -file /usr/ca/root/ca-cert -storepass mldnjava
keytool -keystore /usr/ca/client/client.keystore.jks -alias kafka-single -import -file /usr/ca/client/client.cert-signed -storepass mldnjava
##server.properties
listeners=SSL://kafka-single:9093
advertised.listeners=SSL://kafka-single:9093
ssl.keystore.location=/usr/ca/server/server.keystore.jks
ssl.keystore.password=mldnjava
ssl.key.password=mldnjava
ssl.truststore.location=/usr/ca/trust/server.truststore.jks
ssl.truststore.password=mldnjava
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS
security.inter.broker.protocol=SSL