本文以Java SDK为例介绍如何在VPC环境下使用SDK接入消息队列Kafka版的默认接入点并收发消息。
安装Java依赖库
在pom.xml中添加以下依赖。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.2.2</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.6</version>
</dependency>
说明 建议您保持服务端和客户端版本一致,即保持客户端库版本和消息队列Kafka版实例的大版本一致。您可以在消息队列Kafka版控制台的实例详情页面获取消息队列Kafka版实例的大版本。
准备配置
发送消息
订阅消息
选择以下任意一种方式订阅消息。
- 单Consumer订阅消息。
- 创建单Consumer订阅消息程序KafkaConsumerDemo.java。
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; 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")); //两次Poll之间的最大允许间隔。 //消费者超过该值没有返回心跳,服务端判断消费者处于非存活状态,服务端将消费者从Consumer Group移除并触发Rebalance,默认30s。 props.put(ConsumerConfig.SESSION_TIMEOUT_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 org.apache.kafka.clients.consumer.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) { try { Thread.sleep(1000); } catch (Throwable ignore) { } e.printStackTrace(); } } } }
- 编译并运行KafkaConsumerDemo.java消费消息。
- 创建单Consumer订阅消息程序KafkaConsumerDemo.java。
- 多Consumer订阅消息。
- 创建多Consumer订阅消息程序KafkaMultiConsumerDemo.java。
import java.util.ArrayList; import java.util.List; import java.util.Properties; import java.util.concurrent.atomic.AtomicBoolean; 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.errors.WakeupException; /** * 本教程演示如何在一个进程内开启多个Consumer同时消费Topic。 * 注意全局Consumer数量不要超过订阅的Topic总分区数。 */ public class KafkaMultiConsumerDemo { public static void main(String args[]) throws InterruptedException { //加载kafka.properties。 Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties(); Properties props = new Properties(); //设置接入点,请通过控制台获取对应Topic的接入点。 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers")); //两次Poll之间的最大允许间隔。 //消费者超过该值没有返回心跳,服务端判断消费者处于非存活状态,服务端将消费者从Consumer Group移除并触发Rebalance,默认30s。 props.put(ConsumerConfig.SESSION_TIMEOUT_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")); 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之前消费完这些数据, 且总耗时不得超过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) { //如果关闭则忽略异常。 if (!closed.get()) { throw e; } } finally { consumer.close(); } } //可以被另一个线程调用的关闭Hook。 public void shutdown() { closed.set(true); consumer.wakeup(); } } }
- 编译并运行KafkaMultiConsumerDemo.java消费消息。
- 创建多Consumer订阅消息程序KafkaMultiConsumerDemo.java。