Using SDK to Receive/Send Message (Recommended)

Last updated: 2021-07-14 15:54:38


    This document describes how to access CKafka to receive/send messages with the SDK for Java in a VPC.

    If you need to use SDKs for other languages, please see SDK Documentation.



    Step 1. Add the Java dependent library

    Add the following Java dependent library information to the pom.xml file:


    Step 2. Prepare configurations

    1. Create the CKafka configuration file
    ## Configure the accessed network by copying the information in the **Network** column in the **Access Mode** section on the **Instance Details** page in the console
    ## Configure the topic by copying the information on the **Topic Management** page in the console
    ## Configure the consumer group as needed
    Parameter Description
    bootstrap.servers Accessed network, which can be copied from the Network column in the Access Mode section on the Instance Details page in the console.
    topic Topic name, which can be copied from the Topic Management page in the console. You can customize it. After the demo runs successfully, you can see the consumer on the Consumer Group page.
    1. Create the configuration file loading program
    public class CKafkaConfigurer {
      private static Properties properties;
      public synchronized static Properties getCKafkaProperties() {
         if (null != properties) {
             return properties;
         // Get the content of the configuration file ``
         Properties kafkaProperties = new Properties();
         try {
         } catch (Exception e) {
             System.out.println("getCKafkaProperties error");
         properties = kafkaProperties;
         return kafkaProperties;

    Step 3. Send a message

    1. Write the message production program
    public class CKafkaProducerDemo {
       public static void main(String args[]) {
          // Load ``
          Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();
           Properties properties = new Properties();
          // Set the access point of the corresponding topic, which can be obtained in the console
          properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
           // Set the method for serializing Kafka messages. `StringSerializer` is used in this demo.
          // Set the maximum time to wait for a request
          properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
          // Set the number of retries for the client
          properties.put(ProducerConfig.RETRIES_CONFIG, 5);
          // Set the internal retry interval for the client
          properties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
          // Construct a producer object
          KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
           // Construct a CKafka message
          String topic = kafkaProperties.getProperty("topic"); // Topic of the message. Enter the topic you created in the console
          String value = "this is ckafka msg value"; // Message content
           try {
              // Batch getting future objects can speed up the process. Note that the batch size should not be too large
              List<Future<RecordMetadata>> futureList = new ArrayList<>(128);
              for (int i = 0; i < 10; i++) {
                  // Send the message and get a future object
                  ProducerRecord<String, String> kafkaMsg = new ProducerRecord<>(topic,
                          value + ": " + i);
                  Future<RecordMetadata> metadataFuture = producer.send(kafkaMsg);
              for (Future<RecordMetadata> future : futureList) {
                  // Sync the obtained future object
                  RecordMetadata recordMetadata = future.get();
                  System.out.println("produce send ok: " + recordMetadata.toString());
          } catch (Exception e) {
              // If the sending still fails after client internal retries, the system needs to report and handle the error.
              System.out.println("error occurred");
    1. Compile and run to send the message.
    2. View the execution result.
    Produce ok:ckafka-topic-demo-0@198
    Produce ok:ckafka-topic-demo-0@199
    1. On the Topic Management page in the CKafka console, select the corresponding topic and click More > Message Query and view the just sent message.

    Step 4. Consume the message

    1. Create the subscribed message consumer program
    public class CKafkaConsumerDemo {
       public static void main(String args[]) {
          // Load ``
          Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();
           Properties props = new Properties();
          // Set the access point of the corresponding topic, which can be obtained in the console
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
          // Set the maximum interval between two polls
          // If the consumer does not return a heartbeat message within the interval, the broker determines that the consumer is not alive. the broker removes the consumer from the consumer group and triggers rebalancing. The default value is 30s.
          props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
          // Set the maximum number of messages that can be polled at a time
          // Do not set this parameter to an excessively large value. If polled messages are not all consumed before the next poll starts, load balancing is triggered and lagging occurs.
          props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
          // Set the method for deserializing messages.
          // The instances in the same consumer group consume messages in load balancing mode
          props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty(""));
          // Construct a consumer object. This generates a consumer instance
          KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
          // Set one or more topics to which the consumer group subscribes
          // We recommend you configure consumer instances with the same `GROUP_ID_CONFIG` value to subscribe to the same topics
          List<String> subscribedTopics = new ArrayList<>();
          // If you want to subscribe to multiple topics, add the topics here
          // You must create the topics in the console in advance
          String topicStr = kafkaProperties.getProperty("topic");
          String[] topics = topicStr.split(",");
          for (String topic : topics) {
           // Consume messages in loop
          while (true) {
              try {
                  ConsumerRecords<String, String> records = consumer.poll(1000);
                  // All messages must be consumed before the next poll, and the total duration cannot exceed the timeout interval specified by `SESSION_TIMEOUT_MS_CONFIG`
                  // We recommend you create an independent thread to consume messages and then return the result in async mode
                  for (ConsumerRecord<String, String> record : records) {
                              String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
              } catch (Exception e) {
                  System.out.println("consumer error!");
    1. Compile and run to consume messages.
    2. View the execution result.
    Consume partition:0 offset:298
    Consume partition:0 offset:299
    1. On the Consumer Group page in the CKafka console, select the corresponding consumer group, enter the topic name in Topic Name, and click Query Details to view the consumption details.