基本概念

Kafka TLS双向加密包含的知识涉及到对称加密，非对称加密，Kafka,CA,数字证书。
本文采用C++ librdkafka 库来实现，client 相应证书为AWS(private CA)签名证书

非对称加密

非对称加密算法需要两个密钥：公开密钥（publickey:简称公钥）和私有密钥（privatekey:简称私钥）。公钥与私钥是一对，如果用公钥对数据进行加密，只有用对应的私钥才能解密，反之亦然。

CA

CA，Catificate Authority，它的作用就是提供证书（即服务器证书，由域名、公司信息、序列号和签名信息组成）加强服务端和客户端之间信息交互的安全性，以及证书运维相关服务。
注意，TLS双向加密中CA机构是两个，分别为client CA 和 server CA

TLS双向加密过程

kafka 集群创建及client 公私钥创建参考 https://docs.aws.amazon.com/zh_cn/msk/latest/developerguide/msk-authentication.html
http://kafka.apache.org/documentation/#zk_authz_new_mtls
由于 librdkafka 库需要的参数证书及私钥是分开的，故需要将 jks 文件转换为两个 pem 文件
jks 文件转换为 pem 文件参考https://blog.csdn.net/x7057385/article/details/87162688

TLS 双向认证流程

注意：最后client、server端通信采用的是对称加密通信，对称密钥是经过了加密的，安全性高。

代码

#include <cppkafka/cppkafka.h>
#include <librdkafka/rdkafkacpp.h>

class Producer
{
public:
    Producer()
    {
        cppkafka::Configuration config = {
            { "metadata.broker.list", "localhost:9092" },
            { "security.protocol", "ssl" },
            { "ssl.ca.location", "/home/kafka/client_tls/tls-ca-bundle.pem" },  // public ca,linux系统文件位置为 /etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem
            { "ssl.certificate.location", "/home/kafka/client_tls/cert.pem" },
            { "ssl.key.location", "/home/kafka/client_tls/pri_key.pem" },
            { "ssl.key.password", "123456" },
        };

        producer_ = std::make_shared<cppkafka::Producer>(config);
    }

    virtual ~Producer()
    {
    }

protected:
    std::shared_ptr<cppkafka::Producer> producer_;
};