腾讯云消息队列 CKafka购买和使用

腾讯云消息队列CKafka是一个拥有分布式的、高吞吐量、可扩展性，支持日志收集、监控数据聚合、流式数据处理、在线和离线分析的消息队列服务。
腾讯云消息队列 CKafka开通购买地址 https://cloud.tencent.com/product/ckafka
腾讯云消息队列 CKafka最新官方入门文档 https://cloud.tencent.com/product/ckafka/getting-started
 
腾讯云消息队列CKafka针对开源社区的 CKafka 提供全托管服务，可选的存储和带宽资源，使得腾讯云消息队列CKafka根据需求申请购买，只需专注于业务开发，不用考虑部署运维，低成本、更弹性、更可靠。无论您的业务部署在云上还是云下，用户在控制台创建队列和消费组，即可通过客户端生产与消费消息

本文旨在以最简单明了的方式引导您快速上手消息队列CKafka，为进一步熟悉和使用消息队列 CKafka 的功能提供入门。快速接入消息队列 CKafka 服务的步骤如下：

腾讯云消息队列CKafka购买
登录腾讯云，选择消息队列CKafka，进入消息队列CKafka平台，点击立即购买

创建腾讯云消息队列CKafka实例
1、用户点击立即购买，页面跳转至购买CKafka实例的页面，选择不同规格资源后点击确认完成支付，用户需要依次选择流量峰值、存储容量。如果想清除已选配置，请点击清除配置
2、选择完规格后，点击确认订单，显示确认订单界面界面，查看已选配置是否正确，确认后，点击去支付按钮。在支付界面选择支付类型，支付成功后，跳转到支付成功界面。如果余额不足，请点击去充值
 

 
腾讯云消息队列界面包括集群ID/名称，运行状态，实例类型，付费类型，创建时间，操作。可以按照集群ID/名称进行搜索，创建失败的实例用户可自由选择删除该列表
续费
续费入口：只有创建成功且运行状态为“服务中”的实例，才有允许续费，点击对应实例的续费功能

Topic管理
在实例列表界面选择一个实例，单击名称，进入实例详情界面，选择Topic管理列表，进入Topic管理界面，显示信息包括：Topic名称、分区数、副本数、消息保留时长（小时）、同步落盘、同步复制、操作。Topic可以按照Topic名称进行搜索

创建 Topic
在实例列表界面选择一个实例，单击名称，进入实例详情界面，选择Topic管理。单击新建，进入Topic创建界面
 
 
 

 
点击确定按钮，提示创建Topic成功，查看Topic创建情况
 
删除 Topic
在实例列表界面选择一个实例，单击名称，进入实例详情界面，选择Topic管理。选择对应Topic名称行的删除功能，删除单个Topic。勾选多个Topic、或全选按钮，删除多个Topic
 
连接CKafka集群
创建主机连接CKafka
登录腾讯云系统，选择计算-弹性云主机，点击新建。选择与创建消息队列CKafka一致的区域创建主机，选择对应规格，镜像系统与CKafka系统一致，目前为centos 7.3即可，可选挂载存储。点击确认订单，返回后查看主机状态，选择控制台，进入主机，登录命令窗口，下载CKafka客户端，执行CKafka命令，命令中IP对应为CKafka集群中实例详情界面的IP
 
生产消费Demo演示
通过maven引入jar包依赖
 
maven依赖
 
<dependencies>
  <!-- https://mvnrepository.com/artifact/org.apache.CKafka/CKafka -->
  <dependency>
      <groupId>org.apache.CKafka</groupId>
      <artifactId>CKafka_2.11</artifactId>
      <version>0.10.0.0</version>
  </dependency>
</dependencies>
写公共配置类
 
BigDataConfig
 
public class BigDataConfig {
    /**
    * CKafka 的 topic
    */
    public final static String TOPIC="string112";
 
    /**
    * CKafka broker list
    * 此处用来修改CKafka 的连接地址
    */
 
    public final static String BROKER_URL = "10.0.44.65:9092";
    /**
    * 消费组Id
    */
    public final static String GROUP_ID="tj-group22";
    /**
    * 客户连接 Id
    * 对数据不产生影响，对CKafka而言，只关注 groupId
    */
    public final static String CLIENT_ID="TJclient2";
    /**
    *  CKafka message->key
    *  对生产消费消息，没有影响，可暂时忽略
    */
 
    public final static String KEY="generator";
}
生产者逻辑
 
Producer
 
public class Producer extends Thread {
    private final CKafkaProducer<Integer, String> producer;
    private final String topic;
    private final Boolean isAsync;
 
    public Producer(String topic, Boolean isAsync) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BigDataConfig.BROKER_URL);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producer = new CKafkaProducer<>(props);
        this.topic = topic;
        this.isAsync = isAsync;
    }
public void run() {
        int messageNo = 1;
        while (true) {
            String messageStr = "Message_" + messageNo;
            long startTime = System.currentTimeMillis();
            if (isAsync) {
                producer.send(new ProducerRecord<>(topic,
                        messageNo,
                        messageStr), new DemoCallBack(startTime, messageNo, messageStr));
            } else {
                try {
                    producer.send(new ProducerRecord<>(topic,
                            messageNo,
                            messageStr)).get();
                    System.out.println("Sent message to topic [ "+ topic+ " ] -----> (" + messageNo + ", " + messageStr + ")");
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
            ++messageNo;
        }
    }
}
 
class DemoCallBack implements Callback {
 
    private final long startTime;
    private final int key;
    private final String message;
 
    public DemoCallBack(long startTime, int key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }
 
    /**
    * 消息发送成功后的回调方法
    */
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            String topic = metadata.topic();
            System.out.println(
                    "Producer send to topic 【 "+topic+" 】  message ------>(key: " + key + ",  value: " + message + " ) sent to partition(" + metadata.partition() +
                            "), " +
                            "offset( " + metadata.offset() + " ) 耗时  " + elapsedTime + " 毫秒");
        } else {
            exception.printStackTrace();
        }
    }
}
消费者逻辑
 
Consumer
 
public class Consumer extends ShutdownableThread {
    private final CKafkaConsumer<Integer, String> consumer;
    private final String topic;
 
    public Consumer(String topic) {
        super("CKafkaConsumerExample", false);
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BigDataConfig.BROKER_URL);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, BigDataConfig.GROUP_ID);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.CKafka.common.serialization.IntegerDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.CKafka.common.serialization.StringDeserializer");
 
        consumer = new CKafkaConsumer<>(props);
        this.topic = topic;
    }
 
    @Override
    public void doWork() {
        consumer.subscribe(Collections.singletonList(this.topic));
        ConsumerRecords<Integer, String> records = consumer.poll(1000);
        for (ConsumerRecord<Integer, String> record : records) {
            System.out.println("From topic 【 "+ topic+" 】 Received message------> (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
        }
    }
 
    @Override
    public String name() {
        return null;
    }
 
    @Override
    public boolean isInterruptible() {
        return false;
    }
}
测试类
 
ApplicationForMessage
 
public class ApplicationForMessage {
    public static void main(String[] args) {
        /**
        * 生产者生产消息
        */
        boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync");
        Producer producerThread = new Producer(BigDataConfig.TOPIC, isAsync);
        producerThread.start();
 
        /**
        * 消费者消费消息
        */
        Consumer consumerThread = new Consumer(BigDataConfig.TOPIC);
        consumerThread.start();
    }
}