设置账号 CAM 验证

{
    "statement": [
        {
            "action": [
                "cam:BuildDataFlowAuthToken"
            ],
            "effect": "allow",
            "resource": [
                "qcs::cam::uin/<用户 uin>:resourceUser/<实例 ID>/<账号名>"
            ]
        }
    ],
    "version": "2.0"
}

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringSerializer;
import com.tencentcloudapi.common.Credential;

import java.time.Duration;
import java.util.Properties;

public class CamProducerExample {

    public static void main(String[] args) {
        // ====== 配置参数 ======
        String bootstrapServers = "your-ckafka-server:9092";  // CKafka 接入点
        String topic = "your-topic";                           // Topic 名称
        String saslMechanism = "SCRAM-SHA-256";               // SASL 机制
        String username = "your-username";                     // SASL 用户名
        String region = "ap-guangzhou";                        // 实例所在地域
        String instanceId = "ckafka-xxxxx";                   // CKafka 实例 ID

        // 腾讯云凭证（建议通过环境变量获取）
        String secretId = System.getenv("TENCENTCLOUD_SECRET_ID");
        String secretKey = System.getenv("TENCENTCLOUD_SECRET_KEY");
        Credential credential = new Credential(secretId, secretKey);

        // ====== 通过 CAM 获取 Token 作为密码 ======
        String password = CamTokenHelper.getAuthToken(region, instanceId, username, credential);
        if (password.isEmpty()) {
            System.err.println("获取 CAM Token 失败，退出");
            return;
        }

        // ====== 构建 Kafka Producer 配置 ======
        Properties config = new Properties();
        config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        config.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
        config.put(SaslConfigs.SASL_JAAS_CONFIG, String.format(
            "org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"%s\\" password=\\"%s\\";",
            username, password));
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // ====== 发送消息 ======
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "测试消息-" + System.currentTimeMillis());
            RecordMetadata metadata = producer.send(record).get();
            System.out.printf("消息发送成功: topic=%s, partition=%d, offset=%d%n",
                    metadata.topic(), metadata.partition(), metadata.offset());
        } catch (Exception e) {
            System.err.println("发送消息失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
}


import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.tencentcloudapi.common.Credential;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class CamConsumerExample {

    public static void main(String[] args) {
        // ====== 配置参数 ======
        String bootstrapServers = "your-ckafka-server:9092";  // CKafka 接入点
        String topic = "your-topic";                           // Topic 名称
        String saslMechanism = "SCRAM-SHA-256";               // SASL 机制
        String username = "your-username";                     // SASL 用户名
        String groupId = "your-consumer-group";               // 消费者组 ID
        String region = "ap-guangzhou";                        // 实例所在地域
        String instanceId = "ckafka-xxxxx";                   // CKafka 实例 ID

        // 腾讯云凭证（建议通过环境变量获取）
        String secretId = System.getenv("TENCENTCLOUD_SECRET_ID");
        String secretKey = System.getenv("TENCENTCLOUD_SECRET_KEY");
        Credential credential = new Credential(secretId, secretKey);

        // ====== 通过 CAM 获取 Token 作为密码 ======
        String password = CamTokenHelper.getAuthToken(region, instanceId, username, credential);
        if (password.isEmpty()) {
            System.err.println("获取 CAM Token 失败，退出");
            return;
        }

        // ====== 构建 Kafka Consumer 配置 ======
        Properties config = new Properties();
        config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        config.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
        config.put(SaslConfigs.SASL_JAAS_CONFIG, String.format(
            "org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"%s\\" password=\\"%s\\";",
            username, password));
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

        // ====== 消费消息 ======
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config)) {
            consumer.subscribe(Arrays.asList(topic));
            
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("收到消息: topic=%s, partition=%d, offset=%d, value=%s%n",
                            record.topic(), record.partition(), record.offset(), record.value());
                }
            }
        } catch (Exception e) {
            System.err.println("消费消息失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
}