tencent cloud

## Configure the access network and copy the network information from the Network column of the Access Method module on the instance details page in the console.
bootstrap.servers=xx.xx.xx.xx:xxxx
## Configure a topic and copy the topic information on the topic management page in the console.
topic=XXX
## Configure a consumer group. You can customize the settings.
group.id=XXX

public class CKafkaProducerDemo {

    public static void main(String args[]) {
        //Load kafka.properties.
        Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();

        Properties properties = new Properties();
        //Set the access point. You can obtain the access point of the corresponding topic in the console.
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

        //The serialization method of Kafka messages. StringSerializer is used by the demo.
        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");
        //The maximum request waiting time.
        properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        //Set the number of internal retries on the client.
        properties.put(ProducerConfig.RETRIES_CONFIG, 5);
        //Set the interval of internal retries on the client.
        properties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
        //Construct a producer object.
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //Construct a Kafka message.
        String topic = kafkaProperties.getProperty(“topic”); //The topic to which the message belongs. Fill it in here after applying in the console.
        String value = “this is ckafka msg value”; //The content of the message.

        try {
            //Obtaining Future objects in batches can speed up the process. Be careful not to use too large a batch.
            List<Future<RecordMetadata>> futureList = new ArrayList<>(128);
            for (int i = 0; i < 10; i++) {
                //Send a message and obtain a Future object.
                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) {
                //Synchronously obtain the result of the Future object.
                RecordMetadata recordMetadata = future.get();
                System.out.println("produce send ok: " + recordMetadata.toString());
            }
        } catch (Exception e) {
            //After the client retries internally, the sending still fails. The service needs to respond to this type of error.
            System.out.println("error occurred");
        }
    }
}

Produce ok:ckafka-topic-demo-0@198
Produce ok:ckafka-topic-demo-0@199

public class CKafkaConsumerDemo {

    public static void main(String args[]) {
        //Load kafka.properties.
        Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();

        Properties props = new Properties();
        //Set the access point. You can obtain the access point of the corresponding topic in the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        //Set the maximum allowed interval between two polls.
        //The default value is 30 seconds. If a consumer fails to return a heartbeat within the duration, the server determines that the consumer is not alive, removes it from the consumer group, and triggers the rebalance process.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        //Maximum number per poll.
        //Be careful that this value cannot be too large. If too much data is polled and cannot be consumed before the next poll, a load balancing will be triggered, resulting in delays.
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        //Deserialization method of messages.
        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");
        //Consumption instances of the same group will carry consumption messages.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
        //Construct a consumption object, that is, generate a consumer instance.
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //Set the topics subscribed by the consumer group. Multiple topics can be subscribed to.
        //If GROUP_ID_CONFIG is the same, it is recommended to set the subscribed topics as the same.
        List<String> subscribedTopics = new ArrayList<>();
        //If multiple topics need to be subscribed to, add them here.
        //Create each topic in the console first.
        String topicStr = kafkaProperties.getProperty("topic");
        String[] topics = topicStr.split(",");
        for (String topic : topics) {
            subscribedTopics.add(topic.trim());
        }
        consumer.subscribe(subscribedTopics);

        //Consume messages in loops.
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                //The data must be consumed before the next poll, and the total time cannot exceed the value of SESSION_TIMEOUT_MS_CONFIG.
                //It is recommended to open a separate thread pool to consume messages and then return the results asynchronously.
                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!");
            }
        }
    }
}

Consume partition:0 offset:298
Consume partition:0 offset:299