Java

2024-05-09 02:46:55

1.1.1 Java开发环境搭建

开发环境

l Maven

l Apache Maven 2.5及以上版本，可至Maven官方下载页面下载。JDK

Java Development Kit 1.8及以上版本，可至Oracle官方下载页面下载。

l 安装后注意配置JAVA的环境变量。IntelliJ IDEA

获取并安装IntelliJ IDEA，可至IntelliJ IDEA官方网站下载。

操作步骤

1. 下载Demo包kafka-java-demo.zip

下载后解压，有如下文件：

表1 KAFKA Demo文件清单

文件名	路径	说明
JavaKafkaConfigurer.java	.\src\main\java\javaDemo	读取Kafka配置文件。
KafkaConsumerDemo.java	.\src\main\java\javaDemo	消费消息。
KafkaMultiConsumerDemo.java	.\src\main\java\javaDemo	批量消费消息。
KafkaProducerDemo.java	.\src\main\java\javaDemo	生产消息。
kafka.properties	.\src\main\resources	kafka配置参数
pom.xml	.\	maven配置文件，包含Kafka客户端引用。

2. 打开IntelliJ IDEA，导入Demo。Demo是一个Maven构建的Java工程，因此需要配置JDK环境，以及IDEA的Maven插件。

3. 修改Kafka配置信息。

修改kafka.properties修改demo文件中的属性变量

1.1.2 Java客户端接入示例

本文介绍Maven方式引入Kafka客户端，并完成Kafka实例连接以及消息生产与消费的相关示例。如果您需要在IDE中查看Demo具体表现，请查看Java开发环境搭建。

下文所有Kafka的配置信息，如实例连接地址、Topic名称、用户信息等，请参考收集连接信息获取。

Maven中引入Kafka客户端

安装Java依赖库，pom.xml添加以下依赖。

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-clients</artifactId>

</dependency>

</dependency>

配置修改

1. 如果是SSL接入点，需在控制台下载证书。

2. 修改kafka.properties文件。

##==============================通用配置参数==============================

bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx

topic=xxx

group.id=xxx

##=======================以下内容请根据实际情况配置========================

##SSL接入点配置证书路径

ssl.truststore.location=/xxxx/ssl.client.truststore.jks

3. 如果不是SSL接入点，需要将代码中SSL部分注释掉。如果是SSL接入点，需要在代码中修改证书密码（ssl.client.pw.key）。

//证书store的密码，保持不变props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "xxxxxxx");

配置读取

import java.util.Properties;

public class JavaKafkaConfigurer {

private static Properties properties;

public synchronized static Properties getKafkaProperties() {

if (null != properties) {

return properties;

}

//获取配置文件kafka.properties的内容

Properties kafkaProperties = new Properties();

try {

kafkaProperties.load(KafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));

} catch (Exception e) {

//没加载到文件，程序要考虑退出

e.printStackTrace();

}

properties = kafkaProperties;

return kafkaProperties;

}

生产消息

import java.util.ArrayList;

import java.util.List;

import java.util.Properties;

import java.util.concurrent.Future;

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;

import org.apache.kafka.common.config.SslConfigs;

public class KafkaProducerDemo {

public static void main(String args[]) {

//加载kafka.properties

Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

Properties props = new Properties();

//设置接入点，请通过控制台获取对应Topic的接入点

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

//设置SSL根证书的路径，请记得将XXX修改为自己的路径

props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));

//根证书store的密码，保持不变

props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "c24f5210");

//接入协议，目前支持使用SSL协议接入

props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");

//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);

//hostname校验改成空

props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

//构造Producer对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可；

//如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过5个

KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

//构造一个Kafka消息

String topic = kafkaProperties.getProperty("topic"); //消息所属的Topic，请在控制台申请之后，填写在这里

String value = "this is the message's value"; //消息的内容

try {

//批量获取 futures 可以加快速度, 但注意，批量不要太大

List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);

for (int i =0; i < 100; i++) {

//发送消息，并获得一个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对象的结果

try {

RecordMetadata recordMetadata = future.get();

System.out.println("Produce ok:" + recordMetadata.toString());

} catch (Throwable t) {

t.printStackTrace();

}

} catch (Exception e) {

System.out.println("error occurred");

e.printStackTrace();

}

消费消息

import org.apache.kafka.clients.CommonClientConfigs;

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;

import org.apache.kafka.common.config.SslConfigs;

import java.util.ArrayList;

import java.util.List;

import java.util.Properties;

public class KafkaConsumerDemo {

public static void main(String args[]) {

//加载kafka.properties

Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

Properties props = new Properties();

//设置接入点，请通过控制台获取对应Topic的接入点

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

//设置SSL根证书的路径，请记得将XXX修改为自己的路径

props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));

//根证书store的密码，保持不变

props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "c24f5210");

//接入协议，目前支持使用SSL协议接入

props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");

//两次poll之间的最大允许间隔

//可更加实际拉去数据和客户的版本等设置此值，默认30s

props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);

//设置单次拉取的量，走公网访问时，该参数会有较大影响

props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);

props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);

//每次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"));

//hostname校验改成空

props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

//构造消息对象，也即生成一个消费实例

KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);

//设置消费组订阅的Topic，可以订阅多个

//如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样

List<String> subscribedTopics = new ArrayList<String>();

//如果需要订阅多个Topic，则在这里add进去即可

//每个Topic需要先在控制台进行创建

subscribedTopics.add(kafkaProperties.getProperty("topic"));

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) {

try {

Thread.sleep(1000);

} catch (Throwable ignore) {

}

e.printStackTrace();

}

分布式消息服务Kafka

Java

1.1.1 Java开发环境搭建

1.1.2 Java客户端接入示例