VPC网络如何接入腾讯云消息队列CKafka?

2022年7月25日 24点热度 0人点赞 0条评论

步骤1:创建实例

  1. 登录 CKafka 控制台
  2. 在左侧导航栏单击实例列表,单击新建进入实例购买页,根据自身业务需求选择购买信息。
    • 计费模式:包年包月。
    • 规格类型:根据自身业务需求选择标准版或者专业版。
    • Kafka 版本:根据您的业务需求选择 Kafka 版本,可参见 CKafka 版本选择建议
    • 地域:选择和部署客户端的资源相近的地域。
    • 可用区:
      • 标准版:不支持多可用区部署。
      • 专业版:若当前地域支持多可用区部署,则最多可选择2个可用区进行部署。关于跨可用区部署原理介绍请参见 跨可用区部署
    • 产品规格:根据峰值带宽和磁盘容量选择对应的型号。
    • 消息保留:范围在1小时 - 2160小时。
      在磁盘容量不足(即磁盘利用率达到90%)时,将会提前删除旧的消息,以确保服务可用性。
    • 实例名称:购买多个实例时,支持创建实例后缀数字自动升序以及指定模式串功能。具体操作参见 批量连续命名或指定模式串命名
  3. 根据自身业务需求选择合适的私有网络。
    若用户需要接入其他私有网络可参见 添加路由策略 修改路由接入规则。
  4. 单击购买,大约等待3分钟 - 5分钟即可在实例列表页看到创建好的实例。

步骤2:创建 Topic

  1. 登录 CKafka 控制台 。
  2. 实例列表页,单击 步骤1 创建的实例的“ID/名称”,进入实例详情页。
  3. 在实例详情页,单击页面顶部的 Topic 管理,单击新建
  4. 在编辑 Topic 窗口中,选择分区数和副本数等信息。
    VPC网络如何接入腾讯云消息队列CKafka?插图

    • 名称:Topic 名称,输入后无法更改,名称只能包含字母、数字、下划线、“-”和“.”。
    • 分区数:一个物理上分区的概念,一个 Topic 可以包含一个或者多个 partition,CKafka 以 partition 作为消息分配单位。
    • 副本数:partition 的副本个数,用于保障 partition 的高可用,为保障数据可靠性,当前不支持创建单副本 Topic,默认开启2副本。
      副本数也算分区个数,例如客户创建了1个 Topic、6个分区、2个副本,那么分区额度一共用了1 * 6 * 2 = 12个。
    • 标签:设置资源标签,关于标签的详细介绍请参见 标签管理
    • 预设 ACL 策略:可以选择提前设置好的 ACL 策略,关于 ACL 策略详情请参见 配置 ACL 策略
  5. 单击提交完成 Topic 创建。

步骤3:添加 VPC 网络路由

  1. 在 CKafka 实例列表 页面,单击 步骤1 创建的实例的“ID/名称”。
  2. 在实例详情页面,选择接入方式模块中的添加路由策略,新增一条 VPC 网络路由。
    VPC网络如何接入腾讯云消息队列CKafka?插图1
    添加后获得 VPC 网络访问的域名和接口。
    VPC网络如何接入腾讯云消息队列CKafka?插图2

步骤4:收发信息

使用 SDK 收发消息(推荐)

该任务以 Java 客户端为例指导您使用 VPC 网络接入消息队列 CKafka 并收发消息。

其他语言客户端请参见 SDK文档

前提条件

操作步骤

步骤1:准备配置

  1. 将下载下来的 Demo 上传到同一个 VPC 下的Linux服务器,然后登录 linux 服务器,进入 javakafkademo 下的 VPC 目录。
  2. 修改 VPC 工程下的 resources 目录中的 kafka.properties。
    ## 配置接入网络,在控制台的实例详情页面接入方式模块的网络列复制。
    bootstrap.servers=xx.xx.xx.xx:xxxx
    ## 配置Topic,在控制台上topic管理页面复制。
    topic=XXX
    ## 配置Consumer Group,您可以自定义设置
    group.id=XXX
    参数 说明
    bootstrap.servers 接入网络,在控制台的实例详情页面接入方式模块的网络列复制。
    VPC网络如何接入腾讯云消息队列CKafka?插图3
    topic topic名称,您可以在控制台上topic管理页面复制。
    VPC网络如何接入腾讯云消息队列CKafka?插图4
    group.id 您可以自定义设置,demo运行成功后可以在Consumer Group页面看到该消费者组。

步骤2:发送消息

  1. 编译并运行生产消息程序 CKafkaProducerDemo.java。
    public class CKafkaProducerDemo {
    
        public static void main(String args[]) {
            //加载kafka.properties。
            Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();
    
            Properties properties = new Properties();
            //设置接入点,请通过控制台获取对应Topic的接入点。
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
    
            //消息队列Kafka版消息的序列化方式, 此处demo 使用的是StringSerializer。
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            //请求的最长等待时间。
            properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
            //设置客户端内部重试次数。
            properties.put(ProducerConfig.RETRIES_CONFIG, 5);
            //设置客户端内部重试间隔。
            properties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
            //构造Producer对象。
            KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
    
            //构造一个消息队列Kafka版消息。
            String topic = kafkaProperties.getProperty("topic"); //消息所属的Topic,请在控制台申请之后,填写在这里。
            String value = "this is ckafka msg value"; //消息的内容。
    
            try {
                //批量获取Future对象可以加快速度, 但注意, 批量不要太大。
                List<Future<RecordMetadata>> futureList = new ArrayList<>(128);
                for (int i = 0; i < 10; i++) {
                    //发送消息,并获得一个Future对象。
                    ProducerRecord<String, String> kafkaMsg = new ProducerRecord<>(topic,
                            value + ": " + i);
                    Future<RecordMetadata> metadataFuture = producer.send(kafkaMsg);
                    futureList.add(metadataFuture);
    
                }
                producer.flush();
                for (Future<RecordMetadata> future : futureList) {
                    //同步获得Future对象的结果。
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("produce send ok: " + recordMetadata.toString());
                }
            } catch (Exception e) {
                //客户端内部重试之后,仍然发送失败,业务要应对此类错误。
                System.out.println("error occurred");
            }
        }
    }
  2. 运行结果。
    Produce ok:ckafka-topic-demo-0@198
    Produce ok:ckafka-topic-demo-0@199
  3. 在 CKafka 控制台 的topic管理页面,选择对应的 topic ,单击更多 > 消息查询,查看刚刚发送的消息。
    VPC网络如何接入腾讯云消息队列CKafka?插图5

步骤3:消费消息

  1. 编译并运行 Consumer 订阅消息程序 CKafkaConsumerDemo.java。
    public class CKafkaConsumerDemo {
    
        public static void main(String args[]) {
            //加载kafka.properties。
            Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();
    
            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 KafkaConsumer<>(props);
            //设置消费组订阅的Topic,可以订阅多个。
            //如果GROUP_ID_CONFIG是一样,则订阅的Topic也建议设置成一样。
            List<String> subscribedTopics = new ArrayList<>();
            //如果需要订阅多个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) {
                    System.out.println("consumer error!");
                }
            }
        }
    }
  2. 运行结果。
    Consume partition:0 offset:298
    Consume partition:0 offset:299
  3. 在 CKafka 控制台 的Consumer Group页面,选择对应的消费组名称,在主题名称输入 topic 名称,单击查询详情,查看消费详情。
    VPC网络如何接入腾讯云消息队列CKafka?插图6

运行 Kafka 客户端(可选)

该任务指导您在购买 CKafka 服务后,使用 Kafka API。在腾讯云服务器上搭建 CKafka 环境后,本地下载并解压 Kafka 工具包,并对 Kafka API 进行简单测试。

操作步骤

步骤1:安装 JDK 环境

1. 检查 Java 安装。

打开终端,执行如下命令:

java -version

如果输出 Java 版本号,说明 Java 安装成功;如果没有安装 Java,请 下载安装 Java 软件开发套件(JDK)

2. 设置 Java 环境。

设置 JAVA_HOME 环境变量,并指向您机器上的 Java 安装目录。
以 Java JDK 1.8.0_20 版本为例,操作系统的输出如下:

操作系统 输出
Windows Set the environment variable JAVA_HOME to
C:\Program Files\Java\jdkjdk1.8.0_20
Linux export JAVA_HOME=/usr/local/java-current
Mac OSX export JAVA_HOME=/Library/Java/Home

将 Java 编译器地址添加到系统路径中:

操作系统 输出
Windows 将字符串“;C:\Program Files\Java\jdk1.8.0_20\bin”添加到系统变量“Path”的末尾
Linux export PATH=$PATH:$JAVA_HOME/bin/
Mac OSX not required

使用上面提到的 java -version 命令验证 Java 安装。

步骤2:下载 Kafka 工具包

下载并解压 Kafka 安装包。(Kafka 安装包官网下载地址

步骤3:Kafka API 测试

VPC网络如何接入腾讯云消息队列CKafka?插图7
通过 CLI 命令生产和消费消息,去到./bin目录下。

  1. 打开终端启动消费者。
    bash kafka-console-consumer.sh --bootstrap-server XXXX:port --topic XXXX --consumer.config ../config/consumer.properties

    说明
    • 将 XXXX:port 替换成 VPC 网络访问的域名与端口,在控制台实例详情页面的接入方式模块获取。
      VPC网络如何接入腾讯云消息队列CKafka?插图7
    • topic:将 XXXX 替换成 topic 名称,在控制台 topic 管理页面获取。
  2. 另外开一个终端窗口启动生产者。
    bash kafka-console-producer.sh --broker-list XXXX:port --topic XXXX --producer.config ../config/producer.properties

    说明
    • 将 XXXX:port 替换成 VPC 网络访问的域名与端口,在控制台实例详情页面的接入方式模块获取。
      VPC网络如何接入腾讯云消息队列CKafka?插图7
    • topic:将 XXXX 替换成 topic 名称,在控制台 topic 管理页面获取。

    输入消息内容之后按回车,即可看到消费端也几乎同时收到消息。 生产消息: VPC网络如何接入腾讯云消息队列CKafka?插图8 消费消息: VPC网络如何接入腾讯云消息队列CKafka?插图9

  3. 在 CKafka 控制台消息查询页面,查询刚刚发送的消息内容。
    VPC网络如何接入腾讯云消息队列CKafka?插图5
    消息详情如下:
    VPC网络如何接入腾讯云消息队列CKafka?插图10

zdc1228@163.com

这个人很懒,什么都没留下

文章评论