tencent cloud

VPC Network Access
Download PDF
Last updated: 2026-01-05 15:16:59
VPC Network Access
Last updated: 2026-01-05 15:16:59
Download PDF

Scenarios

This document uses the Go client as an example to describe how to access TDMQ for CKafka (CKafka) via the Virtual Private Cloud (VPC) network and send and receive messages.

Prerequisites

You have installed Go.
You have downloaded the demo.
You have obtained the client connection parameters as instructed in SDK Overview.

Operation Steps

Step 1: Preparing for Configurations

1. Upload gokafkademo from the downloaded demo to the Linux server.
2. Log in to the Linux server, go to the gokafkademo directory, and run the following command to add dependency libraries.
go get -v gopkg.in/confluentinc/confluent-kafka-go.v1/kafka
3. Modify the configuration file kafka.json.
{
  "topic": [
      "test"
  ],
  "bootstrapServers": [
      "xx.xx.xx.xx:xxxx"
  ],
  "consumerGroupId": "yourConsumerId"
}
Parameter
Description
topic
Topic name. Copy the name on the Topic List page in the console.
bootstrapServers
Access network. On the Basic Info page of the instance in the console, select the Access Mode module and copy the network information from the Network column.
consumerGroupId
You can define the name and see the consumer on the Consumer Group page after successful demo running.

Step 2: Sending Messages

1. Write a message production program.
package main
import (
                "fmt"
                "gokafkademo/config"
                "log"
                "strings"	
                "github.com/confluentinc/confluent-kafka-go/kafka"
)
func main() {
   cfg, err := config.ParseConfig("../config/kafka.json")
       if err != nil {
           log.Fatal(err)
       }
   p, err := kafka.NewProducer(&kafka.ConfigMap{
   // Set the access point. You can obtain the access point of the corresponding topic in the console.
   "bootstrap.servers": strings.Join(cfg.Servers, ","),
   // The default value 1 is used if the configuration is not displayed. You can set it based on your business requirements.
   "acks": 1,
   // The number of retries when a request error occurs. It is recommended to set this value to greater than 0 to ensure that the message is not lost to the maximum extent during failed retries.
   "retries": 0,
   // The time between the failed request transmission and the next retry request.
   "retry.backoff.ms": 100,
   // The timeout period for producer network requests.
   "socket.timeout.ms": 6000,
   // Set the interval of internal retries on the client.
   "reconnect.backoff.max.ms": 3000,
       })
       if err != nil {
           log.Fatal(err)
       }
   defer p.Close()
   // Send the generated message to the report handler.
       go func() {
           for e := range p.Events() {
               switch ev := e.(type) {
               case *kafka.Message:
                   if ev.TopicPartition.Error != nil {
                       fmt.Printf("Delivery failed: %v\\n", ev.TopicPartition)
                 } else {
                 fmt.Printf("Delivered message to %v\\n",ev.TopicPartition)
                   }
               }
           }
       }()
   // Send messages asynchronously.
       topic := cfg.Topic[0]
       for _, word := range []string{"Confluent-Kafka", "Golang Client Message"} {
           _ = p.Produce(&kafka.Message{
               TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: 		kafka.PartitionAny},
               Value:          []byte(word),
           }, nil)
       }
   // Wait for messages to be sent.
       p.Flush(10 * 1000)
}
2. Compile and run the program to send messages.
go run main.go
3. View the running results. An example is as follows.
Delivered message to test[0]@628
Delivered message to test[0]@629
4. On the Topic List page in the CKafka console, select the target topic, and choose More > Message Query to view the message just sent.

Step 3: Consuming Messages

1. Write a message consumption program.
package main

import (
    "fmt"
    "gokafkademo/config"
    "log"
    "strings"

    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {

    cfg, err := config.ParseConfig("../config/kafka.json")
    if err != nil {
        log.Fatal(err)
    }

    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        // Set the access point. You can obtain the access point of the corresponding topic in the console.
        "bootstrap.servers": strings.Join(cfg.Servers, ","),
        // The message consumer group that has been set.
        "group.id":          cfg.ConsumerGroupId,
        "auto.offset.reset": "earliest",

        // Consumer timeout interval when the Kafka consumer group mechanism is used. If the broker does not receive the heartbeat from the consumer within this interval, the consumer is considered to be failed, and the broker
        // initiates the rebalancing process again. Currently, the value must be between the value (6000) of the broker configuration parameter group.min.session.timeout.ms and the value (300000) of group.max.session.timeout.ms.
        "session.timeout.ms": 10000,
    })

    if err != nil {
        log.Fatal(err)
    }
    // List of subscribed message topics.
    err = c.SubscribeTopics(cfg.Topic, nil)
    if err != nil {
        log.Fatal(err)
    }

    for {
        msg, err := c.ReadMessage(-1)
        if err == nil {
            fmt.Printf("Message on %s: %s\\n", msg.TopicPartition, string(msg.Value))
        } else {
            // The client will automatically try to clear all errors.
            fmt.Printf("Consumer error: %v (%v)\\n", err, msg)
        }
    }

    c.Close()
}
2. Compile and run the program to consume messages.
go run main.go
3. View the running results. An example is as follows.
Message on test[0]@628: Confluent-Kafka
Message on test[0]@629: Golang Client Message
4. On the Consumer Group page in the CKafka console, select the target consumer group name, enter the topic name in the Topic Name area, and click View Details to view consumption details.
Was this page helpful?
You can also Contact Sales or Submit a Ticket for help.
Yes
No

Feedback