使用實例接入點收發(fā)消息
當(dāng)使用應(yīng)用程序與Kafka集群建立連接時,可通過實例接入點接入云消息隊列 Kafka 版并收發(fā)消息。Kafka實例提供默認接入點、SSL接入點和SASL接入點,以滿足您不同的連接和安全需求。默認接入點適用于在保密性較高的VPC環(huán)境收發(fā)消息;SASL接入點適用于無需對傳輸鏈路加密但需對消息收發(fā)鑒權(quán);同時需要鏈路加密和消息鑒權(quán),建議使用SSL接入點。
環(huán)境準(zhǔn)備
安裝1.8或以上版本JDK。具體操作。請參見安裝JDK。
安裝2.5或以上版本Maven。具體操作,請參見安裝Maven。
安裝編譯工具。
本文以IntelliJ IDEA Ultimate為例。
購買并部署云消息隊列 Kafka 版實例,如果實例類型為VPC實例,則僅顯示默認接入點。如果實例類型為公網(wǎng)/VPC實例,則同時顯示默認接入點和SSL接入點。實例默認不開啟SASL接入點,因此默認不顯示SASL接入點。如需使用SASL接入點,您需要手動開啟。具體操作,請參見SASL用戶授權(quán)。
默認接入點:適用于在VPC環(huán)境收發(fā)消息,但不支持SASL校驗。
SASL接入點:適用于在VPC環(huán)境收發(fā)消息,且支持SASL校驗。
SSL接入點:適用于在公網(wǎng)環(huán)境收發(fā)消息,且支持SASL校驗。
安裝Java依賴庫
使用Java SDK接入Kafka實例的依賴如下所示。此依賴已在下文提到的kafka-java-demo文件夾中的pom.xml內(nèi)置,無需手動添加。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.6</version>
</dependency>
建議您保持服務(wù)端和客戶端版本一致,即保持客戶端庫版本和云消息隊列 Kafka 版實例的大版本一致。您可以在云消息隊列 Kafka 版控制臺的實例詳情頁面獲取云消息隊列 Kafka 版實例的大版本。
準(zhǔn)備配置
可選:下載SSL根證書。如果是SSL接入點,需下載該證書。
訪問Aliware-kafka-demos,單擊,下載Demo工程到本地并解壓。
在解壓的Demo工程中,找到kafka-java-demo文件夾,將此文件夾導(dǎo)入IntelliJ IDEA。
可選:如果采用SSL接入點或者SASL接入點接入實例,需修改配置文件kafka_client_jaas.conf。關(guān)于實例不同接入點的信息,請參見接入點對比。
KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="xxxx" password="xxxx"; };
當(dāng)實例類型為VPC實例時,只有在同一VPC內(nèi)的資源才能訪問Kafka實例,數(shù)據(jù)傳輸具有較高的安全性和私密性。在安全要求更高的場景下,可以開啟ACL功能,消息需結(jié)合SASL身份驗證鑒權(quán)之后再在安全通道傳輸。您可以根據(jù)消息傳輸?shù)陌踩蠹墑e選擇PLAIN機制或SCRAM機制進行身份認證。更多信息,請參見開啟ACL。
當(dāng)實例類型為公網(wǎng)/VPC實例時,公網(wǎng)環(huán)境必須對消息進行鑒權(quán)與加密,SASL的PLAIN機制必須與SSL一起用作傳輸層,采用SASL_SSL協(xié)議,確保消息在沒有加密的情況下不會在線路上傳輸明文。
示例中的username和password為實例的SASL用戶名和密碼。
如果實例未開啟ACL,對于已開啟公網(wǎng)的實例,您可以在云消息隊列 Kafka 版控制臺的實例詳情頁面配置信息區(qū)域獲取默認用戶的用戶名和密碼。
如果實例已開啟ACL,請確保要使用的SASL用戶為PLAIN類型且已授權(quán)收發(fā)消息的權(quán)限,詳情請參見SASL用戶授權(quán)。
修改配置文件kafka.properties。
##==============================通用配置參數(shù)============================== bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx topic=xxx group.id=xxx ##=======================以下內(nèi)容請根據(jù)實際情況配置======================== ##SSL接入點配置 ssl.truststore.location=/xxxx/kafka.client.truststore.jks java.security.auth.login.config=/xxxx/kafka_client_jaas.conf ##SASL接入點PLAIN機制配置 java.security.auth.login.config.plain=/xxxx/kafka_client_jaas_plain.conf ##SASL接入點SCRAM機制配置 java.security.auth.login.config.scram=/xxxx/kafka_client_jaas_scram.conf
參數(shù)
描述
bootstrap.servers
接入點信息。您可在云消息隊列 Kafka 版控制臺的實例詳情頁面的接入點信息區(qū)域獲取。
topic
實例的Topic名稱。您可在云消息隊列 Kafka 版控制臺的Topic 管理頁面獲取。
group.id
實例的Group。您可在云消息隊列 Kafka 版控制臺的Group 管理頁面獲取。
說明如果應(yīng)用運行producer.go發(fā)送消息,該參數(shù)可以不配置;如果應(yīng)用運行consumer.go訂閱消息,該參數(shù)必須配置。
ssl.truststore.location
SSL根證書的路徑。請將上文準(zhǔn)備配置中第一步下載的SSL證書文件保存至本地路徑,用本地路徑替換示例代碼中的xxxx。例如:/home/ssl/kafka.client.truststore.jks。
重要如果是默認接入點實例或者SASL接入點實例,該參數(shù)不需要配置;如果是SSL接入點實例,該參數(shù)必須配置。
java.security.auth.login.config
JAAS配置文件所在路徑,將Demo工程中的kafka_client_jaas.conf文件另存至本地路徑,用本地路徑替換示例代碼中的xxxx。例如:/home/ssl/kafka_client_jaas.conf。
重要如果是默認接入點實例,該參數(shù)不需要配置;如果是SSL接入點實例或者SASL接入點實例,該參數(shù)必須配置。
發(fā)送消息
編譯并運行KafkaProducerDemo.java發(fā)送消息。
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
//如果是SSL接入點實例或者SASL接入點實例,請注釋以下第一行代碼。
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/*
*如果是SSL接入點實例或者SASL接入點實例,請取消以下兩行代碼的注釋。
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/
public class KafkaProducerDemo {
public static void main(String args[]) {
/*
* 如果是SSL接入點實例,請取消以下一行代碼的注釋。
設(shè)置JAAS配置文件的路徑。
JavaKafkaConfigurer.configureSasl();
*/
/*
* 如果是SASL接入點PLAIN機制實例,請取消以下一行代碼的注釋。
設(shè)置JAAS配置文件的路徑。
JavaKafkaConfigurer.configureSaslPlain();
*/
/*
* 如果是SASL接入點SCRAM機制實例,請取消以下一行代碼的注釋。
設(shè)置JAAS配置文件的路徑。
JavaKafkaConfigurer.configureSaslScram();
*/
//加載kafka.properties。
Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
//設(shè)置接入點,請通過控制臺獲取對應(yīng)Topic的接入點。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
/*
* 如果是SSL接入點實例,請取消以下四行代碼的注釋。
* 與sasl路徑類似,該文件也不能被打包到j(luò)ar中。
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
* 根證書store的密碼,保持不變。
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
* 接入?yún)f(xié)議,目前支持使用SASL_SSL協(xié)議接入。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
* SASL鑒權(quán)方式,保持不變。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
*/
/*
* 如果是SASL接入點PLAIN機制實例,請取消以下兩行代碼的注釋。
* 接入?yún)f(xié)議。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
* Plain方式。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
*/
/*
* 如果是SASL接入點SCRAM機制實例,請取消以下兩行代碼的注釋。
* 接入?yún)f(xié)議。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
* SCRAM方式。
props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
*/
//云消息隊列 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);
//設(shè)置客戶端內(nèi)部重試次數(shù)。
props.put(ProducerConfig.RETRIES_CONFIG, 5);
//設(shè)置客戶端內(nèi)部重試間隔。
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
/*
* 如果是SSL接入點實例或,請取消以下一行代碼的注釋。
* Hostname校驗改成空。
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
*/
//構(gòu)造Producer對象,注意,該對象是線程安全的,一般來說,一個進程內(nèi)一個Producer對象即可。
//如果想提高性能,可以多構(gòu)造幾個對象,但不要太多,最好不要超過5個。
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
//構(gòu)造一個云消息隊列 Kafka 版消息。
String topic = kafkaProperties.getProperty("topic"); //消息所屬的Topic,請在控制臺申請之后,填寫在這里。
String value = "this is the message's value"; //消息的內(nèi)容。
try {
//批量獲取Future對象可以加快速度,但注意,批量不要太大。
List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
for (int i =0; i < 100; i++) {
//發(fā)送消息,并獲得一個Future對象。
ProducerRecord<String, String> kafkaMessage = new ProducerRecord<String, String>(topic, value + ": " + i);
Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
futures.add(metadataFuture);
}
producer.flush();
for (Future<RecordMetadata> future: futures) {
//同步獲得Future對象的結(jié)果。
try {
RecordMetadata recordMetadata = future.get();
System.out.println("Produce ok:" + recordMetadata.toString());
} catch (Throwable t) {
t.printStackTrace();
}
}
} catch (Exception e) {
//客戶端內(nèi)部重試之后,仍然發(fā)送失敗,業(yè)務(wù)要應(yīng)對此類錯誤。
System.out.println("error occurred");
e.printStackTrace();
}
}
}
訂閱消息
選擇以下任意一種方式訂閱消息。
單Consumer訂閱消息
編譯并運行KafkaConsumerDemo.java發(fā)送消息。
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* 如果是SSL接入點實例,請取消以下三行代碼的注釋;如果是SASL接入點,請取消以下前兩行代碼的注釋。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/
public class KafkaConsumerDemo {
public static void main(String args[]) {
//設(shè)置JAAS配置文件的路徑。
/*
* 如果是SSL接入點實例,請取消以下一行代碼的注釋。
JavaKafkaConfigurer.configureSasl();
*/
/*
* 如果是SASL接入點PLAIN機制實例,請取消以下一行代碼的注釋。
JavaKafkaConfigurer.configureSaslPlain();
*/
/*
* 如果是SASL接入點SCRAM機制實例,請取消以下一行代碼的注釋。
JavaKafkaConfigurer.configureSaslScram();
*/
//加載kafka.properties
Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
//設(shè)置接入點,請通過控制臺獲取對應(yīng)Topic的接入點。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
//如果是SSL接入點實例,請注釋以下第一行代碼。
//可更加實際拉去數(shù)據(jù)和客戶的版本等設(shè)置此值,默認30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
/*
* 如果是SSL接入點實例,請取消以下六行代碼的注釋。
* 設(shè)置SSL根證書的路徑,請記得將XXX修改為自己的路徑。
* 與SASL路徑類似,該文件也不能被打包到j(luò)ar中。
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
* 根證書存儲的密碼,保持不變。
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
* 接入?yún)f(xié)議,目前支持使用SASL_SSL協(xié)議接入。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
* SASL鑒權(quán)方式,保持不變。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
* 兩次Poll之間的最大允許間隔。
* 消費者超過該值沒有返回心跳,服務(wù)端判斷消費者處于非存活狀態(tài),服務(wù)端將消費者從Group移除并觸發(fā)Rebalance,默認30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
* 設(shè)置單次拉取的量,走公網(wǎng)訪問時,該參數(shù)會有較大影響。
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
*/
//如果是SASL接入點PLAIN機制實例,請注釋以下一行代碼。
//可更加實際拉去數(shù)據(jù)和客戶的版本等設(shè)置此值,默認30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
/*
* 如果是SASL接入點PLAIN機制實例,請取消以下三行代碼的注釋。
* 接入?yún)f(xié)議。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
* Plain方式。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
* 兩次Poll之間的最大允許間隔。
* 消費者超過該值沒有返回心跳,服務(wù)端判斷消費者處于非存活狀態(tài),服務(wù)端將消費者從Group移除并觸發(fā)Rebalance,默認30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
*/
//如果是SASL接入點SCRAM機制實例,請注釋以下一行代碼。
//可更加實際拉去數(shù)據(jù)和客戶的版本等設(shè)置此值,默認30s
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
/*
* 如果是SASL接入點SCRAM機制實例,請取消以下四行代碼的注釋。
* 接入?yún)f(xié)議。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
* SCRAM方式。
props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
* 兩次Poll之間的最大允許間隔。
* 消費者超過該值沒有返回心跳,服務(wù)端判斷消費者處于非存活狀態(tài),服務(wù)端將消費者從Group移除并觸發(fā)Rebalance,默認30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
*/
//每次poll的最大數(shù)量。
//注意該值不要改得太大,如果poll太多數(shù)據(jù),而不能在下次poll之前消費完,則會觸發(fā)一次負載均衡,產(chǎn)生卡頓。
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");
//當(dāng)前消費實例所屬的消費組,請在控制臺申請之后填寫。
//屬于同一個組的消費實例,會負載消費消息。
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
//如果是SSL接入點實例,請取消以下一行代碼的注釋。
//Hostname校驗改成空。
//props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
//構(gòu)造消息對象,也即生成一個消費實例。
KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
//設(shè)置消費組訂閱的Topic,可以訂閱多個。
//如果GROUP_ID_CONFIG是一樣,則訂閱的Topic也建議設(shè)置成一樣。
List<String> subscribedTopics = new ArrayList<String>();
//如果是SSL接入點實例,請注釋以下前五行代碼,取消第六行代碼的注釋。
//如果需要訂閱多個Topic,則在這里add進去即可。
//每個Topic需要先在控制臺進行創(chuàng)建。
String topicStr = kafkaProperties.getProperty("topic");
String[] topics = topicStr.split(",");
for (String topic: topics) {
subscribedTopics.add(topic.trim());
}
//subscribedTopics.add(kafkaProperties.getProperty("topic"));
consumer.subscribe(subscribedTopics);
//循環(huán)消費消息。
while (true){
try {
ConsumerRecords<String, String> records = consumer.poll(1000);
//必須在下次poll之前消費完這些數(shù)據(jù), 且總耗時不得超過SESSION_TIMEOUT_MS_CONFIG。
//建議開一個單獨的線程池來消費消息,然后異步返回結(jié)果。
for (ConsumerRecord<String, String> record : records) {
System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
}
} catch (Exception e) {
try {
Thread.sleep(1000);
} catch (Throwable ignore) {
}
e.printStackTrace();
}
}
}
}
多Consumer訂閱消息
編譯并運行KafkaMultiConsumerDemo.java消費消息。
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
//如果是SSL接入點實例或者SASL接入點實例,請取消以下第一行代碼的注釋。
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* 如果是SSL接入點實例,請取消以下前三行代碼的注釋;如果是SASL接入點實例,請取消以下前兩行代碼的注釋。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/
import org.apache.kafka.common.errors.WakeupException;
/**
* 本教程演示如何在一個進程內(nèi)開啟多個Consumer同時消費Topic。
* 注意全局Consumer數(shù)量不要超過訂閱的Topic總分區(qū)數(shù)。
*/
public class KafkaMultiConsumerDemo {
public static void main(String args[]) throws InterruptedException {
//設(shè)置JAAS配置文件的路徑。
/*
* 如果是SSL接入點實例,請取消以下一行代碼的注釋。
JavaKafkaConfigurer.configureSasl();
*/
/*
* 如果是SASL接入點PLAIN機制實例,請取消以下一行代碼的注釋。
JavaKafkaConfigurer.configureSaslPlain();
*/
/*
* 如果是SASL接入點SCRAM機制實例,請取消以下一行代碼的注釋。
JavaKafkaConfigurer.configureSaslScram();
*/
//加載kafka.properties。
Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
//設(shè)置接入點,請通過控制臺獲取對應(yīng)Topic的接入點。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
/*
* 如果是SSL接入點實例,請取消以下四行代碼的注釋。
* 與SASL路徑類似,該文件也不能被打包到JAR中。
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
* 根證書存儲的密碼,保持不變。
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
* 接入?yún)f(xié)議,目前支持使用SASL_SSL協(xié)議接入。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
* SASL鑒權(quán)方式,保持不變。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
*/
/*
* 如果是SASL接入點PLAIN機制實例,請取消以下兩行代碼的注釋。
* 接入?yún)f(xié)議。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
* Plain方式。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
*/
/*
* 如果是SASL接入點SCRAM機制實例,請取消以下兩行代碼的注釋。
* 接入?yún)f(xié)議。
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
* SCRAM方式。
props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
*/
//兩次Poll之間的最大允許間隔。
//消費者超過該值沒有返回心跳,服務(wù)端判斷消費者處于非存活狀態(tài),服務(wù)端將消費者從Group移除并觸發(fā)Rebalance,默認30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
//每次Poll的最大數(shù)量。
//注意該值不要改得太大,如果Poll太多數(shù)據(jù),而不能在下次Poll之前消費完,則會觸發(fā)一次負載均衡,產(chǎn)生卡頓。
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");
//當(dāng)前消費實例所屬的消費組,請在控制臺申請之后填寫。
//屬于同一個組的消費實例,會負載消費消息。
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
/*
* 如果是SSL接入點實例,請取消以下一行代碼的注釋。
* 構(gòu)造消費對象,也即生成一個消費實例。
* Hostname校驗改成空。
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
*/
int consumerNum = 2;
Thread[] consumerThreads = new Thread[consumerNum];
for (int i = 0; i < consumerNum; i++) {
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
List<String> subscribedTopics = new ArrayList<String>();
subscribedTopics.add(kafkaProperties.getProperty("topic"));
consumer.subscribe(subscribedTopics);
KafkaConsumerRunner kafkaConsumerRunner = new KafkaConsumerRunner(consumer);
consumerThreads[i] = new Thread(kafkaConsumerRunner);
}
for (int i = 0; i < consumerNum; i++) {
consumerThreads[i].start();
}
for (int i = 0; i < consumerNum; i++) {
consumerThreads[i].join();
}
}
static class KafkaConsumerRunner implements Runnable {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final KafkaConsumer consumer;
KafkaConsumerRunner(KafkaConsumer consumer) {
this.consumer = consumer;
}
@Override
public void run() {
try {
while (!closed.get()) {
try {
ConsumerRecords<String, String> records = consumer.poll(1000);
//必須在下次Poll之前消費完這些數(shù)據(jù), 且總耗時不得超過SESSION_TIMEOUT_MS_CONFIG。
for (ConsumerRecord<String, String> record : records) {
System.out.println(String.format("Thread:%s Consume partition:%d offset:%d", Thread.currentThread().getName(), record.partition(), record.offset()));
}
} catch (Exception e) {
try {
Thread.sleep(1000);
} catch (Throwable ignore) {
}
e.printStackTrace();
}
}
} catch (WakeupException e) {
//如果關(guān)閉則忽略異常。
if (!closed.get()) {
throw e;
}
} finally {
consumer.close();
}
}
//可以被另一個線程調(diào)用的關(guān)閉Hook。
public void shutdown() {
closed.set(true);
consumer.wakeup();
}
}
}
常見問題
云消息隊列 Kafka 版的SASL_SSL證書怎么配置?
訪問上文準(zhǔn)備配置的第一步中的地址,下載SSL證書到本地路徑,配置Demo工程中kafka.properties配置文件的ssl.truststore.location
參數(shù),即將SSL證書本地路徑配置在此參數(shù)中。
使用Java SDK通過實例接入點收發(fā)消息時,可以綁定自己的SSL證書嗎?
不支持。建議您使用云消息隊列 Kafka 版提供的SSL證書。
相關(guān)文檔
除了使用實例接入點接入實例外,還可使用Spring Cloud框架接入實例并收發(fā)消息,請參見使用Spring Cloud框架收發(fā)消息。
若使用Java SDK不能成功收發(fā)消息,請檢查實例的運行狀態(tài)是否健康,請參見實例運行健康自檢指南。