消息队列 CKafka 版 配置限流

一、前提条件

配置客户端前，请确认已完成以下准备：

1. CKafka 实例已开启 SASL_SSL 类型的 VPC 接入点或公网接入点。

2. 已获取 SASL_SSL 接入地址、用户名和密码。

3. 客户端版本支持 SASL_SSL 协议。

4. 如使用自定义证书，请确认客户端已获取服务端证书对应的 CA 证书链。

5. 如使用双向认证，请确认已准备客户端证书、客户端私钥及对应密码。

二、认证方式与客户端配置要求

为了完成上述配置，你需要登录 SSL 控制台 ，获取所用证书。

注意：CKafka 自定义证书暂不支持按接入域名进行证书域名校验，客户端需关闭 hostname/domain 校验。

三、单向认证配置

单向认证场景下，客户端需信任服务端证书链。

# 信任库路径及密码，用于验证服务端证书ssl.truststore.location=/Users/ckafka/client.truststore.jks #jks文件地址ssl.truststore.password=your_password# SASL 账号密码配置sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\ username="yourinstance#yourusername" \\ password="your_password"; ssl.endpoint.identification.algorithm=

参数说明：

其他语言请参考对应 SDK 文档，详情请参见 SDK 概述。

四、双向认证配置

双向认证场景下，客户端除信任服务端证书链外，还需提供客户端证书和私钥。

# 1. 验证服务端身份 ssl.truststore.location=/Users/ckafka/client.truststore.jks #jks文件地址ssl.truststore.password=your_password# SASL 账号密码配置sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\ username="your_username" \\ password="your_password";# 2. 提供客户端身份 ssl.keystore.location=/data/ssl/client.keystore.jksssl.keystore.password=client_password # 你生成证书时设置的密码ssl.key.password=key_password # 私钥密码# 注意：双向认证时，请确保 ssl.endpoint.identification.algorithm 为空或设为 ""# 避免因域名校验导致的握手失败ssl.endpoint.identification.algorithm=

参数说明：

其他语言请参考对应 SDK 文档，详情请参见 SDK 概述。

五、默认证书配置

如实例使用 CKafka 默认证书，客户端需信任 CKafka 默认 SSL 证书链。如客户端运行环境默认信任对应证书链，通常无需额外配置 truststore。如连接失败，请根据实际证书链配置客户端信任库。

#信任库路径及密码，用于验证服务端证书ssl.truststore.location=/Users/ckafka/client.truststore.jks #jks文件地址ssl.truststore.password=5fi6R!M# SASL 账号密码配置sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\ username="yourinstance#yourusername" \\ password="your_password"; ssl.endpoint.identification.algorithm=

六、配置文件加载实例

public class CKafkaConfigurer { private static Properties properties; public static void configureSaslPlain() { //如果用 -D 或者其它方式设置过，这里不再设置。 if (null == System.getProperty("java.security.auth.login.config")) { //请注意将 XXX 修改为自己的路径。 System.setProperty("java.security.auth.login.config", getCKafkaProperties().getProperty("java.security.auth.login.config.plain")); } } public synchronized static Properties getCKafkaProperties() { if (null != properties) { return properties; } //获取配置文件 kafka.properties 的内容。 Properties kafkaProperties = new Properties(); try { kafkaProperties.load(CKafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties")); } catch (Exception e) { System.out.println("getCKafkaProperties error"); } properties = kafkaProperties; return kafkaProperties; }}

七、Producer 配置示例

public class KafkaSaslProducerDemo { public static void main(String[] args) { //设置 JAAS 配置文件的路径。 CKafkaConfigurer.configureSaslPlain(); //加载 kafka.properties。 Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties(); Properties props = new Properties(); //设置接入点，请通过控制台获取对应 Topic 的接入点。 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers")); // // SASL_SSL 公网接入 // // 接入协议。 props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL"); // SASL 采用 Plain 方式。 props.put(SaslConfigs.SASL_MECHANISM, "PLAIN"); // SSL 加密。 props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)); props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,kafkaProperties.getProperty(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)); //消息队列 Kafka 版消息的序列化方式。 props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); //请求的最长等待时间。 props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000); //设置客户端内部重试次数。 props.put(ProducerConfig.RETRIES_CONFIG, 5); //设置客户端内部重试间隔。 props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000); //ack=0 producer 将不会等待来自 broker 的确认，重试配置不会生效。注意如果被限流了后，就会被关闭连接。 //ack=1 broker leader 将不会等待所有 broker follower 的确认，就返回 ack。 //ack=all broker leader 将等待所有 broker follower 的确认，才返回 ack。 props.put(ProducerConfig.ACKS_CONFIG, "all"); //构造 Producer 对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可。 KafkaProducer<String, String> producer = new KafkaProducer<>(props); //构造一个消息队列 Kafka 版消息。 String topic = kafkaProperties.getProperty("topic"); //消息所属的Topic，请在控制台申请之后，填写在这里。 String value = "this is ckafka msg value"; //消息的内容。 try { //批量获取 Future 对象可以加快速度。但注意，批量不要太大。 List<Future<RecordMetadata>> futures = new ArrayList<>(128); for (int i = 0; i < 100; i++) { //发送消息，并获得一个Future对象。 ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic, value + ": " + i); Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage); futures.add(metadataFuture); } producer.flush(); for (Future<RecordMetadata> future : futures) { //同步获得 Future 对象的结果。 RecordMetadata recordMetadata = future.get(); System.out.println("Produce ok:" + recordMetadata.toString()); } } catch (Exception e) { //客户端内部重试之后，仍然发送失败，业务要应对此类错误。 System.out.println("error occurred"); } }}

八、Consumer 配置示例

public class KafkaSaslConsumerDemo { public static void main(String[] args) { //设置JAAS配置文件的路径。 CKafkaConfigurer.configureSaslPlain(); //加载kafka.properties。 Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties(); Properties props = new Properties(); //设置接入点，请通过控制台获取对应Topic的接入点。 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers")); // // SASL_SSL 公网接入 // // 接入协议。 props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL"); // SASL 采用 Plain 方式。 props.put(SaslConfigs.SASL_MECHANISM, "PLAIN"); // SSL 加密。 props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)); props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,kafkaProperties.getProperty(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)); //消费者超时时长 //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从 Consumer Group 移除并触发Rebalance，默认30s。 props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000); //两次poll的最长时间间隔 //0.10.1.0 版本前这2个概念是混合的，都用session.timeout.ms表示 props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30000); //每次 Poll 的最大数量。 //注意该值不要改得太大，如果 Poll 太多数据，而不能在下次 Poll 之前消费完，则会触发一次负载均衡，产生卡顿。 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30); //消息的反序列化方式。 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); //当前消费实例所属的消费组，请在控制台申请之后填写。 //属于同一个组的消费实例，会负载消费消息。 props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id")); //构造消费对象，也即生成一个消费实例。 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); //设置消费组订阅的 Topic，可以订阅多个。 //如果 GROUP_ID_CONFIG 是一样，则订阅的 Topic 也建议设置成一样。 List<String> subscribedTopics = new ArrayList<String>(); //如果需要订阅多个 Topic，则在这里添加进去即可。 //每个 Topic 需要先在控制台进行创建。 String topicStr = kafkaProperties.getProperty("topic"); String[] topics = topicStr.split(","); for (String topic : topics) { subscribedTopics.add(topic.trim()); } consumer.subscribe(subscribedTopics); //循环消费消息。 while (true) { try { ConsumerRecords<String, String> records = consumer.poll(1000); //必须在下次 Poll 之前消费完这些数据, 且总耗时不得超过 SESSION_TIMEOUT_MS_CONFIG。 for (ConsumerRecord<String, String> record : records) { System.out.println( String.format("Consume partition:%d offset:%d", record.partition(), record.offset())); } } catch (Exception e) { System.out.println("consumer error!"); } } }}