尧图网页设计 
消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载Apache Pulsar 的端到端加密End-to-End Encryption允许应用在生产端加密消息、在消费端解密消息整个加密/解密过程完全发生在客户端侧Broker 只负责存储和转发密文服务端全程不接触任何明文与密钥。本指南围绕 Pulsar 2.3.2 版本的加密机制系统讲解其非对称对称混合加密原理、密钥对的生成、CryptoKeyReader接口的定制实现、Java/C/Python/Node.js 四类客户端的配置方式以及密钥轮换与故障处理策略。读完本文你将能够独立为 Pulsar 应用配置端到端消息加密并理解其底层实现细节。加密模型用对称 AES 加密数据、用非对称密钥保护数据密钥Pulsar 的端到端加密采用典型的混合加密hybrid encryption方案兼顾了对称加密的性能与非对称加密的密钥分发便利性数据加密每条消息使用动态生成的对称 AES 密钥data key加密消息负载payload保证加解密性能数据密钥保护应用提供的ECDSA/RSA 非对称密钥对用于加密这把 AES 数据密钥因此无需把对称密钥分享给所有人。密钥对在这里的职责划分非常明确生产者的密钥是密钥对中的公钥消费者的密钥是密钥对中的私钥。应用将公钥配置给生产者生产者用公钥加密 AES 数据密钥加密后的数据密钥作为消息头message header的一部分随消息一同发送。只有持有私钥的实体即消费者能够解密出数据密钥进而解密消息负载。该设计在源码层面有完整对应客户端 API 定义了MessageCrypto接口见 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageCrypto.java其中encrypt(...)使用数据密钥加密负载并将密钥名与加密后的数据密钥写入消息元数据messageMetadatadecrypt(...)从消息元数据中取出密钥名、用私钥还原数据密钥并解密负载接口常量IV_LEN 12表明底层采用 AES-GCM 模式的 12 字节初始向量IV约定。Pulsar 默认的实现位于 pulsar-client-messagecrypto-bc/src/main/java/org/apache/pulsar/client/impl/crypto/MessageCryptoBc.java基于 Bouncy Castle 加密库并支持通过addPublicKeyCipher向同一数据密钥注册多个公钥密文。一次消息可以用多个密钥加密Pulsar 允许用多个公钥加密同一条消息生产者把多个公钥都注册到当前数据密钥上消息头里会携带多份“加密后的数据密钥”每一份对应一个密钥名。解密时只要消费者持有其中任意一个私钥就能解出数据密钥并还原消息。这在跨应用边界共享消息的场景中非常实用——一个消费者只需要拥有众多接收方中某一方的私钥即可。密钥不存储在 Pulsar 服务端一个必须强调的安全边界是Pulsar 服务本身不存储任何加密密钥。密钥只存在于应用侧。因此如果私钥丢失或被删除对应消息将永久丢失、不可恢复。这要求应用把私钥纳入严谨的密钥管理体系KMS、HSM 或受控的文件存储。加密整体流程Producer / Consumer 视角生产端与消费端的完整加解密链路如下图所示。生产端流程见 site2/website-next/static/assets/pulsar-encryption-producer.jpgPulsar 生产者端加密流程消费端流程见 site2/website-next/static/assets/pulsar-encryption-consumer.jpgPulsar 消费者端解密流程结合两图与源码可以总结出各环节的关键动作生产端Producer客户端库为待发送的Payload动态生成一把对称AESKey用AESKey加密Payload得到Encrypted Payload用应用提供的非对称公钥加密AESKey同时记录客户端密钥名Client key name将Encrypted Payload与“加密后的 AESKey 密钥名”封装为Encrypted message通过 wire 协议发送给 Broker。Broker 全程只接触密文。消费端Consumer从 Broker 收到Encrypted message含Encrypted Payload与Encrypted AESKey Client key name消费者用自身持有的非对称私钥配合消息附带的密钥名解出AESKey用AESKey解密Encrypted Payload还原原始业务数据。快速上手生成密钥对并配置客户端第 1 步用 OpenSSL 生成 ECDSA 密钥对首先在命令行生成 ECDSA或 RSA公/私密钥对。以下命令使用secp521r1曲线生成 ECDSA 私钥并导出公钥openssl ecparam -name secp521r1 -genkey -param_enc explicit -out test_ecdsa_privkey.pem openssl ec -in test_ecdsa_privkey.pem -pubout -outform pem -out test_ecdsa_pubkey.pem生成后你会得到test_ecdsa_privkey.pem私钥与test_ecdsa_pubkey.pem公钥两个文件。若使用 RSA可改用openssl genrsa/openssl rsa -pubout系列命令格式同样为 PEM。第 2 步接入密钥管理体系将生成的公钥与私钥纳入应用的密钥管理体系并让生产者客户端能够获取公钥、消费者客户端能够获取私钥。Pulsar 本身不托管密钥这一步完全由应用自行实现例如对接 KMS、HSM 或本地受控文件。第 3 步实现CryptoKeyReader接口客户端通过CryptoKeyReader接口加载密钥。该接口定义在 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/CryptoKeyReader.java包含两个方法getPublicKey(String keyName, MapString, String metadata)生产者调用返回EncryptionKeyInfo内含公钥字节getPrivateKey(String keyName, MapString, String metadata)消费者调用返回EncryptionKeyInfo内含私钥字节。源码注释中有一个重要提示这两个方法在生产者创建时以及消费者接收消息时都会被调用实现中不应包含阻塞调用否则会阻塞消息收发路径。仓库中已经提供了开箱即用的默认实现 pulsar-client/src/main/java/org/apache/pulsar/client/impl/DefaultCryptoKeyReader.java它从文件系统读取公钥/私钥文件支持通过DefaultCryptoKeyReader.builder()构建。Java 官方教程示例可参考 pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleCryptoProducer.java 与 SampleCryptoConsumer.java。第 4 步为生产者指定加密密钥名在生产者构建器上添加加密密钥名key name客户端会用它从CryptoKeyReader中定位对应的密钥PulsarClient.newProducer().addEncryptionKey(myapp.key);addEncryptionKey可以链式调用多次从而为一条消息配置多个密钥详见下文“多密钥加密”。第 5 步为 Producer / Consumer / Reader 配置CryptoKeyReaderJavaPulsarClient pulsarClient PulsarClient.builder().serviceUrl(pulsar://localhost:6650).build(); String topic persistent://my-tenant/my-ns/my-topic; // RawFileKeyReader 仅为示例实现并非 Pulsar 官方提供生产环境可直接使用 DefaultCryptoKeyReader CryptoKeyReader keyReader new RawFileKeyReader(test_ecdsa_pubkey.pem, test_ecdsa_privkey.pem); Producerbyte[] producer pulsarClient.newProducer() .topic(topic) .cryptoKeyReader(keyReader) .addEncryptionKey(myappkey) .create(); Consumerbyte[] consumer pulsarClient.newConsumer() .topic(topic) .subscriptionName(my-subscriber-name) .cryptoKeyReader(keyReader) .subscribe(); Readerbyte[] reader pulsarClient.newReader() .topic(topic) .startMessageId(MessageId.earliest) .cryptoKeyReader(keyReader) .create();CC 客户端内置了DefaultCryptoKeyReader实现可直接传入公钥/私钥文件路径Client client(pulsar://localhost:6650); std::string topic persistent://my-tenant/my-ns/my-topic; // DefaultCryptoKeyReader 是内置实现从文件读取公钥和私钥 auto keyReader std::make_sharedDefaultCryptoKeyReader(test_ecdsa_pubkey.pem, test_ecdsa_privkey.pem); Producer producer; ProducerConfiguration producerConf; producerConf.setCryptoKeyReader(keyReader); producerConf.addEncryptionKey(myappkey); client.createProducer(topic, producerConf, producer); Consumer consumer; ConsumerConfiguration consumerConf; consumerConf.setCryptoKeyReader(keyReader); client.subscribe(topic, my-subscriber-name, consumerConf, consumer); Reader reader; ReaderConfiguration readerConf; readerConf.setCryptoKeyReader(keyReader); client.createReader(topic, MessageId::earliest(), readerConf, reader);PythonPython 客户端同样内置了从文件读取密钥的CryptoKeyReaderfrom pulsar import Client, CryptoKeyReader client Client(pulsar://localhost:6650) topic persistent://my-tenant/my-ns/my-topic # CryptoKeyReader 是内置实现从文件读取公钥和私钥 key_reader CryptoKeyReader(test_ecdsa_pubkey.pem, test_ecdsa_privkey.pem) producer client.create_producer( topictopic, encryption_keymyappkey, crypto_key_readerkey_reader ) consumer client.subscribe( topictopic, subscription_namemy-subscriber-name, crypto_key_readerkey_reader ) reader client.create_reader( topictopic, start_message_idMessageId.earliest, crypto_key_readerkey_reader ) client.close()Node.JSNode.js 客户端通过publicKeyPath与privateKeyPath直接指定密钥文件路径const Pulsar require(pulsar-client); (async () { // Create a client const client new Pulsar.Client({ serviceUrl: pulsar://localhost:6650, operationTimeoutSeconds: 30, }); // Create a producer const producer await client.createProducer({ topic: persistent://public/default/my-topic, sendTimeoutMs: 30000, batchingEnabled: true, publicKeyPath: public-key.client-rsa.pem, encryptionKey: encryption-key }); // Create a consumer const consumer await client.subscribe({ topic: persistent://public/default/my-topic, subscription: sub1, subscriptionType: Shared, ackTimeoutMs: 10000, privateKeyPath: private-key.client-rsa.pem }); // Send messages for (let i 0; i 10; i 1) { const msg my-message-${i}; producer.send({ data: Buffer.from(msg), }); console.log(Sent message: ${msg}); } await producer.flush(); // Receive messages for (let i 0; i 10; i 1) { const msg await consumer.receive(); console.log(msg.getData().toString()); consumer.acknowledge(msg); } await consumer.close(); await producer.close(); await client.close(); })();第 6 步自定义CryptoKeyReader实现官方并未在客户端中捆绑一个开箱即用的 Java 文件读取实现因此文档给出了一个自定义实现示例也可直接使用仓库中的DefaultCryptoKeyReaderJavaclass RawFileKeyReader implements CryptoKeyReader { String publicKeyFile ; String privateKeyFile ; RawFileKeyReader(String pubKeyFile, String privKeyFile) { publicKeyFile pubKeyFile; privateKeyFile privKeyFile; } Override public EncryptionKeyInfo getPublicKey(String keyName, MapString, String keyMeta) { EncryptionKeyInfo keyInfo new EncryptionKeyInfo(); try { keyInfo.setKey(Files.readAllBytes(Paths.get(publicKeyFile))); } catch (IOException e) { System.out.println(ERROR: Failed to read public key from file publicKeyFile); e.printStackTrace(); } return keyInfo; } Override public EncryptionKeyInfo getPrivateKey(String keyName, MapString, String keyMeta) { EncryptionKeyInfo keyInfo new EncryptionKeyInfo(); try { keyInfo.setKey(Files.readAllBytes(Paths.get(privateKeyFile))); } catch (IOException e) { System.out.println(ERROR: Failed to read private key from file privateKeyFile); e.printStackTrace(); } return keyInfo; } }EncryptionKeyInfo承载密钥字节setKey(...)以及可选的密钥元数据。实现时同样需要注意这两个方法会沿消息收发路径被频繁调用不应包含阻塞操作如网络 RPC 到远端 KMS否则会拖慢生产与消费吞吐。Cclass CustomCryptoKeyReader : public CryptoKeyReader { public: Result getPublicKey(const std::string keyName, std::mapstd::string, std::string metadata, EncryptionKeyInfo encKeyInfo) const override { // TODO: return ResultOk; } Result getPrivateKey(const std::string keyName, std::mapstd::string, std::string metadata, EncryptionKeyInfo encKeyInfo) const override { // TODO: return ResultOk; } }; auto keyReader std::make_sharedCustomCryptoKeyReader(/* ... */); // TODO: create producer, consumer or reader based on keyReader hereC 侧同样可以使用默认实现DefaultCryptoKeyReader只需传入private key与public key的文件路径即可。Python / Node.js截至 Pulsar 2.3.2 版本Python 与 Node.js 客户端暂不支持自定义CryptoKeyReader实现只能使用内置的默认实现通过指定私钥/公钥文件路径。密钥轮换Key RotationPulsar 的密钥轮换机制分为两个层面AES 数据密钥Pulsar 会每 4 小时或在发布一定数量消息后自动生成一把新的 AES 数据密钥。这把对称密钥只存在于内存中用于加密当下时段的消息负载非对称公钥生产者每 4 小时调用一次CryptoKeyReader.getPublicKey()获取最新的公钥版本从而感知密钥对是否轮换。因此当你在密钥管理系统中轮换应用密钥对后生产者最迟会在 4 小时内拿到新公钥并用于后续消息的加密。需要注意的是旧密钥失效后用旧公钥加密的历史消息仍需对应的旧私钥才能解密应用应保留足够长的私钥留档窗口避免“轮换即失密”。在生产端启用多密钥加密当消息会被跨应用边界消费时你需要确保其他应用中的消费者能够拿到至少一把能够解密消息的私钥。有两种典型做法消费方应用把他们的公钥提供给你你将其加入生产者的密钥列表你从生产者使用的密钥对中向对方授予一把私钥的访问权。生产者需要用多个密钥加密消息时把所有这些密钥名加入配置即可。消费者只要持有其中任意一个私钥就能解密。例如需要用myapp.messagekey1与myapp.messagekey2两把密钥加密消息PulsarClient.newProducer().addEncryptionKey(myapp.messagekey1).addEncryptionKey(myapp.messagekey2);这与MessageCrypto.addPublicKeyCipher(SetString keyNames, CryptoKeyReader keyReader)的语义一致同一个数据密钥会用传入的多个公钥分别加密密文随消息头分发。在消费端解密加密消息消费者需要持有生产者加密时所用密钥对中的任意一把私钥才能解密消息。如果你希望某个消费方接收加密消息操作路径是为该消费方生成一对公/私钥把公钥交给生产者应用生产者用这把公钥加密消息消费方则用对应的私钥在本地解密。整个过程不需要在 Pulsar 服务端做任何配置。故障处理Handle Failures生产者/消费者丢失密钥访问权生产者侧加密失败时生产动作会失败并指出失败原因。应用可以调用PulsarClient.newProducer().cryptoFailureAction(ProducerCryptoFailureAction)控制生产者行为。枚举定义于 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerCryptoFailureAction.java仅有两个取值FAIL默认加密操作失败则发送请求失败SEND忽略加密失败继续以明文发送消息。使用SEND意味着消息可能以明文形式落盘请务必评估降级明文发送带来的安全风险。消费者侧若因解密失败或缺少密钥导致消费失败应用可以调用PulsarClient.newConsumer().cryptoFailureAction(ConsumerCryptoFailureAction)控制消费行为。枚举定义于 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerCryptoFailureAction.java包含三个取值FAIL默认消费失败直到加密成功为止DISCARD消息被静默确认ack不会投递给应用CONSUME把密文投递给应用由应用自行负责解密。源码注释明确指出若消息同时被压缩解密将失败若消息包含批消息batch客户端无法取出批内的单条消息。需要特别提醒如果私钥永久丢失应用永远无法解密这些消息无论选择哪种ConsumerCryptoFailureAction。批消息Batch Messaging与积压Backlog批消息如果解密失败且消息是批消息batch客户端无法从批中取出单条消息因此即使把cryptoFailureAction()设置为ConsumerCryptoFailureAction.CONSUME消息消费依然会失败。对应实现可参见 pulsar-client-cpp/lib/BatchMessageContainerBase.h 等批消息容器对加密路径的耦合。积压增长解密失败时消息消费会停止应用除了在客户端日志中看到解密失败记录外还会观察到backlog积压持续增长。如果应用确实拿不到私钥唯一的选择是跳过或丢弃积压中的消息例如通过DISCARD策略或管理手段清理。小结与最佳实践Pulsar 端到端加密把安全边界收敛在客户端侧Broker 永不接触明文与密钥密钥全生命周期由应用管控。实践中的几条核心建议密钥管理是安全性的根基私钥一旦丢失密文永久不可恢复务必接入 KMS/HSM 并做好私钥留档CryptoKeyReader实现必须轻量getPublicKey/getPrivateKey会被生产创建与消费接收路径高频调用避免阻塞调用如需对接远端密钥库可考虑本地缓存 异步刷新配合 4 小时轮换节奏按需选择失败策略生产者默认FAIL如业务允许降级可显式选SEND消费者默认FAILDISCARD适合丢弃不可解密消息CONSUME适合应用自持密钥的解密场景但需注意批消息与压缩消息的限制多密钥协作时提前规划分发跨应用共享消息时用addEncryptionKey聚合多方公钥消费方持任意一把私钥即可解密善用官方示例与测试Java 端的加密读写示例见 pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleCryptoProducer.java 与 SampleCryptoConsumer.java可在接入前快速验证整体链路。更完整的加密使用示例可参考仓库中的 cookbooks-encryption.md其中包含可运行的加密生产/消费代码示例。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 端到端加密End-to-End Encryption完整实战指南混合加密架构、密钥管理与多语言实现Apache Pulsar 端到端加密End to End Encryption完整实战指南混合加密架构、密钥管理与多语言实现 本指南基于 Apache消息队列后端流处理Apache Pulsar 端到端消息加密End-to-End Encryption实战指南混合加密机制、CryptoKeyReader 接入与密钥轮换Apache Pulsar 端到端消息加密End to End Encryption实战指南混合加密机制、CryptoKeyReader 接入与密钥轮换消息队列后端流处理Apache Pulsar 端到端加密End-to-End Encryption完整实战指南对称加密与非对称加密原理、CryptoKeyReader 接口与密钥轮换Apache Pulsar 端到端加密End to End Encryption完整实战指南对称加密与非对称加密原理、CryptoKeyReader 接口消息队列后端流处理上一篇如何快速入门数据工程Data Engineering Wiki新手完全教程 下一篇OpCore Simplify实战指南从零搭建稳定黑苹果系统的智能方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Pulsar 端到端加密（End-to-End Encryption）实战指南：混合加密原理、密钥管理与多语言客户端配置
消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载Apache Pulsar 的端到端加密End-to-End Encryption允许应用在生产端加密消息、在消费端解密消息整个加密/解密过程完全发生在客户端侧Broker 只负责存储和转发密文服务端全程不接触任何明文与密钥。本指南围绕 Pulsar 2.3.2 版本的加密机制系统讲解其非对称对称混合加密原理、密钥对的生成、CryptoKeyReader接口的定制实现、Java/C/Python/Node.js 四类客户端的配置方式以及密钥轮换与故障处理策略。读完本文你将能够独立为 Pulsar 应用配置端到端消息加密并理解其底层实现细节。加密模型用对称 AES 加密数据、用非对称密钥保护数据密钥Pulsar 的端到端加密采用典型的混合加密hybrid encryption方案兼顾了对称加密的性能与非对称加密的密钥分发便利性数据加密每条消息使用动态生成的对称 AES 密钥data key加密消息负载payload保证加解密性能数据密钥保护应用提供的ECDSA/RSA 非对称密钥对用于加密这把 AES 数据密钥因此无需把对称密钥分享给所有人。密钥对在这里的职责划分非常明确生产者的密钥是密钥对中的公钥消费者的密钥是密钥对中的私钥。应用将公钥配置给生产者生产者用公钥加密 AES 数据密钥加密后的数据密钥作为消息头message header的一部分随消息一同发送。只有持有私钥的实体即消费者能够解密出数据密钥进而解密消息负载。该设计在源码层面有完整对应客户端 API 定义了MessageCrypto接口见 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageCrypto.java其中encrypt(...)使用数据密钥加密负载并将密钥名与加密后的数据密钥写入消息元数据messageMetadatadecrypt(...)从消息元数据中取出密钥名、用私钥还原数据密钥并解密负载接口常量IV_LEN 12表明底层采用 AES-GCM 模式的 12 字节初始向量IV约定。Pulsar 默认的实现位于 pulsar-client-messagecrypto-bc/src/main/java/org/apache/pulsar/client/impl/crypto/MessageCryptoBc.java基于 Bouncy Castle 加密库并支持通过addPublicKeyCipher向同一数据密钥注册多个公钥密文。一次消息可以用多个密钥加密Pulsar 允许用多个公钥加密同一条消息生产者把多个公钥都注册到当前数据密钥上消息头里会携带多份“加密后的数据密钥”每一份对应一个密钥名。解密时只要消费者持有其中任意一个私钥就能解出数据密钥并还原消息。这在跨应用边界共享消息的场景中非常实用——一个消费者只需要拥有众多接收方中某一方的私钥即可。密钥不存储在 Pulsar 服务端一个必须强调的安全边界是Pulsar 服务本身不存储任何加密密钥。密钥只存在于应用侧。因此如果私钥丢失或被删除对应消息将永久丢失、不可恢复。这要求应用把私钥纳入严谨的密钥管理体系KMS、HSM 或受控的文件存储。加密整体流程Producer / Consumer 视角生产端与消费端的完整加解密链路如下图所示。生产端流程见 site2/website-next/static/assets/pulsar-encryption-producer.jpgPulsar 生产者端加密流程消费端流程见 site2/website-next/static/assets/pulsar-encryption-consumer.jpgPulsar 消费者端解密流程结合两图与源码可以总结出各环节的关键动作生产端Producer客户端库为待发送的Payload动态生成一把对称AESKey用AESKey加密Payload得到Encrypted Payload用应用提供的非对称公钥加密AESKey同时记录客户端密钥名Client key name将Encrypted Payload与“加密后的 AESKey 密钥名”封装为Encrypted message通过 wire 协议发送给 Broker。Broker 全程只接触密文。消费端Consumer从 Broker 收到Encrypted message含Encrypted Payload与Encrypted AESKey Client key name消费者用自身持有的非对称私钥配合消息附带的密钥名解出AESKey用AESKey解密Encrypted Payload还原原始业务数据。快速上手生成密钥对并配置客户端第 1 步用 OpenSSL 生成 ECDSA 密钥对首先在命令行生成 ECDSA或 RSA公/私密钥对。以下命令使用secp521r1曲线生成 ECDSA 私钥并导出公钥openssl ecparam -name secp521r1 -genkey -param_enc explicit -out test_ecdsa_privkey.pem openssl ec -in test_ecdsa_privkey.pem -pubout -outform pem -out test_ecdsa_pubkey.pem生成后你会得到test_ecdsa_privkey.pem私钥与test_ecdsa_pubkey.pem公钥两个文件。若使用 RSA可改用openssl genrsa/openssl rsa -pubout系列命令格式同样为 PEM。第 2 步接入密钥管理体系将生成的公钥与私钥纳入应用的密钥管理体系并让生产者客户端能够获取公钥、消费者客户端能够获取私钥。Pulsar 本身不托管密钥这一步完全由应用自行实现例如对接 KMS、HSM 或本地受控文件。第 3 步实现CryptoKeyReader接口客户端通过CryptoKeyReader接口加载密钥。该接口定义在 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/CryptoKeyReader.java包含两个方法getPublicKey(String keyName, MapString, String metadata)生产者调用返回EncryptionKeyInfo内含公钥字节getPrivateKey(String keyName, MapString, String metadata)消费者调用返回EncryptionKeyInfo内含私钥字节。源码注释中有一个重要提示这两个方法在生产者创建时以及消费者接收消息时都会被调用实现中不应包含阻塞调用否则会阻塞消息收发路径。仓库中已经提供了开箱即用的默认实现 pulsar-client/src/main/java/org/apache/pulsar/client/impl/DefaultCryptoKeyReader.java它从文件系统读取公钥/私钥文件支持通过DefaultCryptoKeyReader.builder()构建。Java 官方教程示例可参考 pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleCryptoProducer.java 与 SampleCryptoConsumer.java。第 4 步为生产者指定加密密钥名在生产者构建器上添加加密密钥名key name客户端会用它从CryptoKeyReader中定位对应的密钥PulsarClient.newProducer().addEncryptionKey(myapp.key);addEncryptionKey可以链式调用多次从而为一条消息配置多个密钥详见下文“多密钥加密”。第 5 步为 Producer / Consumer / Reader 配置CryptoKeyReaderJavaPulsarClient pulsarClient PulsarClient.builder().serviceUrl(pulsar://localhost:6650).build(); String topic persistent://my-tenant/my-ns/my-topic; // RawFileKeyReader 仅为示例实现并非 Pulsar 官方提供生产环境可直接使用 DefaultCryptoKeyReader CryptoKeyReader keyReader new RawFileKeyReader(test_ecdsa_pubkey.pem, test_ecdsa_privkey.pem); Producerbyte[] producer pulsarClient.newProducer() .topic(topic) .cryptoKeyReader(keyReader) .addEncryptionKey(myappkey) .create(); Consumerbyte[] consumer pulsarClient.newConsumer() .topic(topic) .subscriptionName(my-subscriber-name) .cryptoKeyReader(keyReader) .subscribe(); Readerbyte[] reader pulsarClient.newReader() .topic(topic) .startMessageId(MessageId.earliest) .cryptoKeyReader(keyReader) .create();CC 客户端内置了DefaultCryptoKeyReader实现可直接传入公钥/私钥文件路径Client client(pulsar://localhost:6650); std::string topic persistent://my-tenant/my-ns/my-topic; // DefaultCryptoKeyReader 是内置实现从文件读取公钥和私钥 auto keyReader std::make_sharedDefaultCryptoKeyReader(test_ecdsa_pubkey.pem, test_ecdsa_privkey.pem); Producer producer; ProducerConfiguration producerConf; producerConf.setCryptoKeyReader(keyReader); producerConf.addEncryptionKey(myappkey); client.createProducer(topic, producerConf, producer); Consumer consumer; ConsumerConfiguration consumerConf; consumerConf.setCryptoKeyReader(keyReader); client.subscribe(topic, my-subscriber-name, consumerConf, consumer); Reader reader; ReaderConfiguration readerConf; readerConf.setCryptoKeyReader(keyReader); client.createReader(topic, MessageId::earliest(), readerConf, reader);PythonPython 客户端同样内置了从文件读取密钥的CryptoKeyReaderfrom pulsar import Client, CryptoKeyReader client Client(pulsar://localhost:6650) topic persistent://my-tenant/my-ns/my-topic # CryptoKeyReader 是内置实现从文件读取公钥和私钥 key_reader CryptoKeyReader(test_ecdsa_pubkey.pem, test_ecdsa_privkey.pem) producer client.create_producer( topictopic, encryption_keymyappkey, crypto_key_readerkey_reader ) consumer client.subscribe( topictopic, subscription_namemy-subscriber-name, crypto_key_readerkey_reader ) reader client.create_reader( topictopic, start_message_idMessageId.earliest, crypto_key_readerkey_reader ) client.close()Node.JSNode.js 客户端通过publicKeyPath与privateKeyPath直接指定密钥文件路径const Pulsar require(pulsar-client); (async () { // Create a client const client new Pulsar.Client({ serviceUrl: pulsar://localhost:6650, operationTimeoutSeconds: 30, }); // Create a producer const producer await client.createProducer({ topic: persistent://public/default/my-topic, sendTimeoutMs: 30000, batchingEnabled: true, publicKeyPath: public-key.client-rsa.pem, encryptionKey: encryption-key }); // Create a consumer const consumer await client.subscribe({ topic: persistent://public/default/my-topic, subscription: sub1, subscriptionType: Shared, ackTimeoutMs: 10000, privateKeyPath: private-key.client-rsa.pem }); // Send messages for (let i 0; i 10; i 1) { const msg my-message-${i}; producer.send({ data: Buffer.from(msg), }); console.log(Sent message: ${msg}); } await producer.flush(); // Receive messages for (let i 0; i 10; i 1) { const msg await consumer.receive(); console.log(msg.getData().toString()); consumer.acknowledge(msg); } await consumer.close(); await producer.close(); await client.close(); })();第 6 步自定义CryptoKeyReader实现官方并未在客户端中捆绑一个开箱即用的 Java 文件读取实现因此文档给出了一个自定义实现示例也可直接使用仓库中的DefaultCryptoKeyReaderJavaclass RawFileKeyReader implements CryptoKeyReader { String publicKeyFile ; String privateKeyFile ; RawFileKeyReader(String pubKeyFile, String privKeyFile) { publicKeyFile pubKeyFile; privateKeyFile privKeyFile; } Override public EncryptionKeyInfo getPublicKey(String keyName, MapString, String keyMeta) { EncryptionKeyInfo keyInfo new EncryptionKeyInfo(); try { keyInfo.setKey(Files.readAllBytes(Paths.get(publicKeyFile))); } catch (IOException e) { System.out.println(ERROR: Failed to read public key from file publicKeyFile); e.printStackTrace(); } return keyInfo; } Override public EncryptionKeyInfo getPrivateKey(String keyName, MapString, String keyMeta) { EncryptionKeyInfo keyInfo new EncryptionKeyInfo(); try { keyInfo.setKey(Files.readAllBytes(Paths.get(privateKeyFile))); } catch (IOException e) { System.out.println(ERROR: Failed to read private key from file privateKeyFile); e.printStackTrace(); } return keyInfo; } }EncryptionKeyInfo承载密钥字节setKey(...)以及可选的密钥元数据。实现时同样需要注意这两个方法会沿消息收发路径被频繁调用不应包含阻塞操作如网络 RPC 到远端 KMS否则会拖慢生产与消费吞吐。Cclass CustomCryptoKeyReader : public CryptoKeyReader { public: Result getPublicKey(const std::string keyName, std::mapstd::string, std::string metadata, EncryptionKeyInfo encKeyInfo) const override { // TODO: return ResultOk; } Result getPrivateKey(const std::string keyName, std::mapstd::string, std::string metadata, EncryptionKeyInfo encKeyInfo) const override { // TODO: return ResultOk; } }; auto keyReader std::make_sharedCustomCryptoKeyReader(/* ... */); // TODO: create producer, consumer or reader based on keyReader hereC 侧同样可以使用默认实现DefaultCryptoKeyReader只需传入private key与public key的文件路径即可。Python / Node.js截至 Pulsar 2.3.2 版本Python 与 Node.js 客户端暂不支持自定义CryptoKeyReader实现只能使用内置的默认实现通过指定私钥/公钥文件路径。密钥轮换Key RotationPulsar 的密钥轮换机制分为两个层面AES 数据密钥Pulsar 会每 4 小时或在发布一定数量消息后自动生成一把新的 AES 数据密钥。这把对称密钥只存在于内存中用于加密当下时段的消息负载非对称公钥生产者每 4 小时调用一次CryptoKeyReader.getPublicKey()获取最新的公钥版本从而感知密钥对是否轮换。因此当你在密钥管理系统中轮换应用密钥对后生产者最迟会在 4 小时内拿到新公钥并用于后续消息的加密。需要注意的是旧密钥失效后用旧公钥加密的历史消息仍需对应的旧私钥才能解密应用应保留足够长的私钥留档窗口避免“轮换即失密”。在生产端启用多密钥加密当消息会被跨应用边界消费时你需要确保其他应用中的消费者能够拿到至少一把能够解密消息的私钥。有两种典型做法消费方应用把他们的公钥提供给你你将其加入生产者的密钥列表你从生产者使用的密钥对中向对方授予一把私钥的访问权。生产者需要用多个密钥加密消息时把所有这些密钥名加入配置即可。消费者只要持有其中任意一个私钥就能解密。例如需要用myapp.messagekey1与myapp.messagekey2两把密钥加密消息PulsarClient.newProducer().addEncryptionKey(myapp.messagekey1).addEncryptionKey(myapp.messagekey2);这与MessageCrypto.addPublicKeyCipher(SetString keyNames, CryptoKeyReader keyReader)的语义一致同一个数据密钥会用传入的多个公钥分别加密密文随消息头分发。在消费端解密加密消息消费者需要持有生产者加密时所用密钥对中的任意一把私钥才能解密消息。如果你希望某个消费方接收加密消息操作路径是为该消费方生成一对公/私钥把公钥交给生产者应用生产者用这把公钥加密消息消费方则用对应的私钥在本地解密。整个过程不需要在 Pulsar 服务端做任何配置。故障处理Handle Failures生产者/消费者丢失密钥访问权生产者侧加密失败时生产动作会失败并指出失败原因。应用可以调用PulsarClient.newProducer().cryptoFailureAction(ProducerCryptoFailureAction)控制生产者行为。枚举定义于 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerCryptoFailureAction.java仅有两个取值FAIL默认加密操作失败则发送请求失败SEND忽略加密失败继续以明文发送消息。使用SEND意味着消息可能以明文形式落盘请务必评估降级明文发送带来的安全风险。消费者侧若因解密失败或缺少密钥导致消费失败应用可以调用PulsarClient.newConsumer().cryptoFailureAction(ConsumerCryptoFailureAction)控制消费行为。枚举定义于 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerCryptoFailureAction.java包含三个取值FAIL默认消费失败直到加密成功为止DISCARD消息被静默确认ack不会投递给应用CONSUME把密文投递给应用由应用自行负责解密。源码注释明确指出若消息同时被压缩解密将失败若消息包含批消息batch客户端无法取出批内的单条消息。需要特别提醒如果私钥永久丢失应用永远无法解密这些消息无论选择哪种ConsumerCryptoFailureAction。批消息Batch Messaging与积压Backlog批消息如果解密失败且消息是批消息batch客户端无法从批中取出单条消息因此即使把cryptoFailureAction()设置为ConsumerCryptoFailureAction.CONSUME消息消费依然会失败。对应实现可参见 pulsar-client-cpp/lib/BatchMessageContainerBase.h 等批消息容器对加密路径的耦合。积压增长解密失败时消息消费会停止应用除了在客户端日志中看到解密失败记录外还会观察到backlog积压持续增长。如果应用确实拿不到私钥唯一的选择是跳过或丢弃积压中的消息例如通过DISCARD策略或管理手段清理。小结与最佳实践Pulsar 端到端加密把安全边界收敛在客户端侧Broker 永不接触明文与密钥密钥全生命周期由应用管控。实践中的几条核心建议密钥管理是安全性的根基私钥一旦丢失密文永久不可恢复务必接入 KMS/HSM 并做好私钥留档CryptoKeyReader实现必须轻量getPublicKey/getPrivateKey会被生产创建与消费接收路径高频调用避免阻塞调用如需对接远端密钥库可考虑本地缓存 异步刷新配合 4 小时轮换节奏按需选择失败策略生产者默认FAIL如业务允许降级可显式选SEND消费者默认FAILDISCARD适合丢弃不可解密消息CONSUME适合应用自持密钥的解密场景但需注意批消息与压缩消息的限制多密钥协作时提前规划分发跨应用共享消息时用addEncryptionKey聚合多方公钥消费方持任意一把私钥即可解密善用官方示例与测试Java 端的加密读写示例见 pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleCryptoProducer.java 与 SampleCryptoConsumer.java可在接入前快速验证整体链路。更完整的加密使用示例可参考仓库中的 cookbooks-encryption.md其中包含可运行的加密生产/消费代码示例。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 端到端加密End-to-End Encryption完整实战指南混合加密架构、密钥管理与多语言实现Apache Pulsar 端到端加密End to End Encryption完整实战指南混合加密架构、密钥管理与多语言实现 本指南基于 Apache消息队列后端流处理Apache Pulsar 端到端消息加密End-to-End Encryption实战指南混合加密机制、CryptoKeyReader 接入与密钥轮换Apache Pulsar 端到端消息加密End to End Encryption实战指南混合加密机制、CryptoKeyReader 接入与密钥轮换消息队列后端流处理Apache Pulsar 端到端加密End-to-End Encryption完整实战指南对称加密与非对称加密原理、CryptoKeyReader 接口与密钥轮换Apache Pulsar 端到端加密End to End Encryption完整实战指南对称加密与非对称加密原理、CryptoKeyReader 接口消息队列后端流处理上一篇如何快速入门数据工程Data Engineering Wiki新手完全教程 下一篇OpCore Simplify实战指南从零搭建稳定黑苹果系统的智能方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考