全部产品

VPC 接入

更新时间:2019-03-27 15:04:01

步骤一:准备配置

1. 添加 Maven 依赖

添加 Maven 依赖的示例代码如下:

  1. // 消息队列 Kafka 服务端版本是 0.10.0.0,客户端建议使用该版本
  2. <dependency>
  3. <groupId>org.apache.kafka</groupId>
  4. <artifactId>kafka-clients</artifactId>
  5. <version>0.10.0.0</version>
  6. </dependency>

2. 客户端配置

2.1 准备配置文件:kafka.properties

请参考以下示例代码进行修改:

  1. ## 配置接入点,即控制台的实例详情页显示的“默认接入点”
  2. bootstrap.servers=192.168.0.1:9092,192.168.0.2:9092
  3. ## 配置 Topic,可以在控制台上创建 Topic
  4. topic=alikafka-topic-demo
  5. ## 配置 Consumer Grouo,可以在控制台创建 Consumer Group
  6. group.id=CID-consumer-group-demo

2.2 加载配置文件

请参考以下示例代码进行修改:

  1. public class JavaKafkaConfigurer {
  2. private static Properties properties;
  3. public synchronized static Properties getKafkaProperties() {
  4. if (null != properties) {
  5. return properties;
  6. }
  7. //获取配置文件 kafka.properties 的内容
  8. Properties kafkaProperties = new Properties();
  9. try {
  10. kafkaProperties.load(KafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));
  11. } catch (Exception e) {
  12. // 没加载到文件,程序要考虑退出
  13. e.printStackTrace();
  14. }
  15. properties = kafkaProperties;
  16. return kafkaProperties;
  17. }
  18. }

步骤二:使用 Java SDK 发送消息

说明:本节已 Java SDK 为例,关于其他语言示例,请参见消息队列 Kafka Demo 库

示例代码如下:

  1. // 加载 kafka.properties
  2. Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
  3. Properties props = new Properties();
  4. // 设置接入点,即控制台的实例详情页显示的“默认接入点”
  5. props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
  6. // 接入协议
  7. props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
  8. // Kafka 消息的序列化方式
  9. props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  10. props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  11. // 请求的最长等待时间
  12. props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
  13. // 构造 Producer 对象,注意,该对象是线程安全的,一般来说,一个进程内一个 Producer 对象即可;
  14. // 如果想提高性能,可以多构造几个对象,但不要太多,最好不要超过 5 个
  15. KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
  16. // 构造一个 Kafka 消息
  17. String topic = kafkaProperties.getProperty("topic"); //消息所属的 Topic,请在控制台申请之后,填写在这里
  18. String value = "this is the message's value"; //消息的内容
  19. ProducerRecord<String, String> kafkaMessage = new ProducerRecord<String, String>(topic, value);
  20. try {
  21. // 发送消息,并获得一个 Future 对象
  22. Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
  23. // 同步获得 Future 对象的结果
  24. RecordMetadata recordMetadata = metadataFuture.get();
  25. System.out.println("Produce ok:" + recordMetadata.toString());
  26. } catch (Exception e) {
  27. // 要考虑重试
  28. System.out.println("error occurred");
  29. e.printStackTrace();
  30. }

步骤三:使用 Java SDK 订阅消息

说明:本节已 Java SDK 为例,关于其他语言示例,请参见消息队列 Kafka Demo 库

示例代码如下:

  1. //加载kafka.properties
  2. Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
  3. Properties props = new Properties();
  4. //设置接入点,即控制台的实例详情页显示的“默认接入点”
  5. props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
  6. //两次poll之间的最大允许间隔
  7. //请不要改得太大,服务器会掐掉空闲连接,不要超过30000
  8. props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 25000);
  9. //每次poll的最大数量
  10. //注意该值不要改得太大,如果poll太多数据,而不能在下次poll之前消费完,则会触发一次负载均衡,产生卡顿
  11. props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
  12. //消息的反序列化方式
  13. props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  14. props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  15. //当前消费实例所属的消费组,请在控制台申请之后填写
  16. //属于同一个组的消费实例,会负载消费消息
  17. props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
  18. //构造消息对象,也即生成一个消费实例
  19. KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
  20. //设置消费组订阅的Topic,可以订阅多个
  21. //如果GROUP_ID_CONFIG是一样,则订阅的Topic也建议设置成一样
  22. List<String> subscribedTopics = new ArrayList<String>();
  23. //如果需要订阅多个Topic,则在这里add进去即可
  24. //每个Topic需要先在控制台进行创建
  25. subscribedTopics.add(kafkaProperties.getProperty("topic"));
  26. consumer.subscribe(subscribedTopics);
  27. //循环消费消息
  28. while (true){
  29. try {
  30. ConsumerRecords<String, String> records = consumer.poll(1000);
  31. //必须在下次poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG
  32. //建议开一个单独的线程池来消费消息,然后异步返回结果
  33. for (ConsumerRecord<String, String> record : records) {
  34. System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
  35. }
  36. } catch (Exception e) {
  37. try {
  38. Thread.sleep(1000);
  39. } catch (Throwable ignore) {
  40. }
  41. //参考常见报错: https://help.aliyun.com/document_detail/68168.html?spm=a2c4g.11186623.6.567.2OMgCB
  42. e.printStackTrace();
  43. }
  44. }