步骤二:使用SDK收发消息

最近更新时间:2020-01-21 11:29:12

VPC内访问接入

1)在客户端主机上配置/etc/hosts文件 需要将master1上的/etc/hosts文件内容添加至客户端主机上。
2)添加maven依赖

// 消息队列 Kafka 版服务端版本为 0.10.1.0,建议的客户端版本为 0.10.2.2
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
<version>0.10.2.2</version>
</dependency>

3)客户端配置 配置文件:

##配置节点列表,即控制台中集群详情页中,节点配置项所有节点的内网ip列表,如172.31.0.5:6667,172.31.0.16:6667,172.31.0.4:6667
bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx
##配置 Topic,可以在Kafka
topic=alikafka-topic-demo 
##配置 Consumer Group
group.id=group-demo

加载配置文件

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;
    }
}

4)使用Java SDK发送消息

// 加载 kafka.properties
Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();
Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
// 接入协议
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
// 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);
// 构造 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"; 
//消息的内容
ProducerRecord<String, String>  kafkaMessage =  new ProducerRecord<String, String>(topic, value);try {
  // 发送消息,并获得一个 Future 对象
  Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
  // 同步获得 Future 对象的结果
  RecordMetadata recordMetadata = metadataFuture.get();
  System.out.println("Produce ok:" + recordMetadata.toString());
} catch (Exception e) {
  // 要考虑重试
  System.out.println("error occurred");
  e.printStackTrace();
}   

5)使用Java SDK订阅消息

    //加载kafka.properties
    Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();
    Properties props = new Properties();
    //设置server
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
    //两次poll之间的最大允许间隔
    //请不要改得太大,服务器会掐掉空闲连接,不要超过30000
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 25000);
    //每次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,则在这里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集群创建时默认只有一个公网IP(绑定在节点master1上),如果需要通过公网进行发布和订阅消息,需要对所有节点进行外网EIP的绑定,方法如下:
1)查看所有节点列表
步骤二:使用SDK收发消息
2)在云服务控制台绑定公网IP
在云服务器控制台根据主机名搜索所有节点,设置弹性EIP详见https://docs.ksyun.com/documents/5375
3)开放安全组
创建集群默认创建了名为KSCKMR 的安全组,需要在该安全组内配置外网访问入站的ACL。开放对应客户端外网IP的访问,端口为6667。
4)参照VPC内访问接入 将bootstrap.servers 配置改为外网EIP即可。

金山云,开启您的云计算之旅

免费注册