tencent cloud

PrevNext

Feedback

search by keyword

Recent Pages

Documentation
TDMQ for Apache Pulsar
    Documentation
    Download PDF

    SDK for Go

    Last updated: 2024-06-28 11:33:56
    Download PDF

      Overview

      This document describes how to use open-source SDK to send and receive messages by using the SDK for Go as an example and helps you better understand the message sending and receiving processes.

      Prerequisites

      You have created the required resources as instructed in Resource Creation and Preparation.
      You have installed Go.
      You have downloaded the demo at here.

      Directions

      1. Import the pulsar-client-go library in the client environment.
      1.1 Run the following command in the client environment to download the dependency package of the Pulsar client.
      go get -u "github.com/apache/pulsar-client-go/pulsar"
      1.2 After the installation is completed, use the following code to import the client into your Go project file.
      import "github.com/apache/pulsar-client-go/pulsar"
      2. Create a Pulsar client.
       // Create a Pulsar client
       client, err := pulsar.NewClient(pulsar.ClientOptions{
           // Service access address
           URL: serviceUrl,
           // Authorize the role token
           Authentication:    pulsar.NewAuthenticationToken(authentication),
           OperationTimeout:  30 * time.Second,
           ConnectionTimeout: 30 * time.Second,
       })
      
       if err != nil{
           log.Fatalf("Could not instantiate Pulsar client: %v", err)
       }
      
       defer client.Close()
      
      Parameter
      Description
      serviceUrl
      Cluster access address, which can be viewed and copied on the Cluster Management page in the console.
      
      Authentication
      Role token, which can be copied in the Token column on the Role Management page.
      
      3. Create a producer.
       // Create a producer with the client
       producer, err := client.CreateProducer(pulsar.ProducerOptions{
           // Complete path of the topic in the format of `persistent://cluster (tenant) ID/namespace/topic name`
           Topic: "persistent://pulsar-mmqwr5xx9n7g/sdk_go/topic1",
       })
      
       if err != nil{
           log.Fatal(err)
       }
       defer producer.Close()
      Note:
      You need to enter the complete path of the topic name, i.e., persistent://clusterid/namespace/Topic, where the clusterid/namespace/topic part can be copied directly from the Topic Management page in the console.
      4. Send a message.
       // Send the message
       _, err = producer.Send(context.Background(), &pulsar.ProducerMessage{
           // Message content
           Payload: []byte("hello go client, this is a message."),
           // Business key
           Key: "yourKey",
           // Business parameter
           Properties: map[string]string{"key": "value"},
       })
      5. Create a consumer.
       // Create a consumer with the client
       consumer, err := client.Subscribe(pulsar.ConsumerOptions{
           // Complete path of the topic in the format of `persistent://cluster (tenant) ID/namespace/topic name`
           Topic:            "persistent://pulsar-mmqwr5xx9n7g/sdk_go/topic1",
           // Subscription name
           SubscriptionName: "topic1_sub",
           // Subscription mode
           Type:             pulsar.Shared,
       })
       if err != nil{
           log.Fatal(err)
       }
       defer consumer.Close()
      Note:
      You need to enter the complete path of the topic name, i.e., persistent://clusterid/namespace/Topic, where the clusterid/namespace/topic part can be copied directly from the Topic Management page in the console.
      img
      
      You need to enter the subscription name in the subscriptionName parameter, which can be viewed on the Consumption Management page.
      6. Consume a message.
       // Obtain the message
       msg, err := consumer.Receive(context.Background())
       if err != nil{
           log.Fatal(err)
       }
       // Simulate business processing
       fmt.Printf("Received message msgId: %#v -- content: '%s'\\n",
                  msg.ID(), string(msg.Payload()))
      
       // If the consumption is successful, return `ack`; otherwise, return `nack` or `ReconsumeLater` according to your business needs
       consumer.Ack(msg)
      7. Log in to the TDMQ for Apache Pulsar console, click Topic > Topic Name to enter the consumption management page, and click the triangle below a subscription name to view the production and consumption records.
      img
      
      
      Note:
      The above is a brief introduction to the way of publishing and subscribing to messages. For more operations, see Demo or Pulsar Go client.

      Customizing Log File Output

      Use Cases

      As many users don’t customize the logging library when using the Pulsar SDK for Go, logs are output to os.Stderr by default, as shown below:
      // It's recommended to make this a global instance called `log`.
      func New() *Logger {
          return &Logger{
              Out:          os.Stderr, // Default output address
              Formatter:    new(TextFormatter),
              Hooks:        make(LevelHooks),
              Level:        InfoLevel,
              ExitFunc:     os.Exit,
              ReportCaller: false,
          }
      }
      Generally, log information is output to os.Stderr. If you don’t specify a custom logging library, the SDK for Go logs and business logs will be mixed, making it difficult for troubleshooting.

      Solution

      With the logger API exposed on the client by the SDK for Go, you can customize the log output format and location and use logging libraries such as logrus and zap. Related parameters are as follows:
      1. Implement the log.Logger API provided by the Pulsar SDK for Go by customizing log lib.
      // ClientOptions is used to construct a Pulsar Client instance.
      type ClientOptions struct {
          // Configure the logger used by the client.
          // By default, a wrapped logrus.StandardLogger will be used, namely,
          // log.NewLoggerWithLogrus(logrus.StandardLogger())
          // FIXME: use `logger` as internal field name instead of `log` as it's more idiomatic
          Logger log.Logger
      }
      When using the SDK for Go, you can customize the logger API to customize log lib so that you can redirect logs to a specified location. Taking logrus as an example, the demo below shows you how to customize log lib to output the SDK for Go logs to a specified file.
      package main
      
      import(
          "fmt"
          "io"
          "os"
      
          "github.com/apache/pulsar-client-go/pulsar/log"
          "github.com/sirupsen/logrus"
      
      )
      
      // logrusWrapper implements Logger interface
      // based on underlying logrus.FieldLogger
      type logrusWrapper struct {
          l logrus.FieldLogger
      }
      
      // NewLoggerWithLogrus creates a new logger which wraps
      // the given logrus.Logger
      func NewLoggerWithLogrus(logger *logrus.Logger, outputPath string) log.Logger {
          writer1 := os.Stdout
          writer2, err := os.OpenFile(outputPath, os.O_WRONLY|os.O_CREATE, 0755)
          if err != nil{
              logrus.Error("create file log.txt failed: %v", err)
          }
          logger.SetOutput(io.MultiWriter(writer1, writer2))
          return &logrusWrapper{
              l: logger,
          }
      }
      
      func (l *logrusWrapper) SubLogger(fs log.Fields) log.Logger {
          return &logrusWrapper{
              l: l.l.WithFields(logrus.Fields(fs)),
          }
      }
      
      func (l *logrusWrapper) WithFields(fs log.Fields) log.Entry {
          return logrusEntry{
              e: l.l.WithFields(logrus.Fields(fs)),
          }
      }
      
      func (l *logrusWrapper) WithField(name string, value interface{}) log.Entry {
          return logrusEntry{
              e: l.l.WithField(name, value),
          }
      }
      
      func (l *logrusWrapper) WithError(err error) log.Entry {
          return logrusEntry{
              e: l.l.WithError(err),
          }
      }
      
      func (l *logrusWrapper) Debug(args ...interface{}) {
          l.l.Debug(args...)
      }
      
      func (l *logrusWrapper) Info(args ...interface{}) {
          l.l.Info(args...)
      }
      
      func (l *logrusWrapper) Warn(args ...interface{}) {
          l.l.Warn(args...)
      }
      
      func (l *logrusWrapper) Error(args ...interface{}) {
          l.l.Error(args...)
      }
      
      func (l *logrusWrapper) Debugf(format string, args ...interface{}) {
          l.l.Debugf(format, args...)
      }
      
      func (l *logrusWrapper) Infof(format string, args ...interface{}) {
          l.l.Infof(format, args...)
      }
      
      func (l *logrusWrapper) Warnf(format string, args ...interface{}) {
          l.l.Warnf(format, args...)
      }
      
      func (l *logrusWrapper) Errorf(format string, args ...interface{}) {
          l.l.Errorf(format, args...)
      }
      
      type logrusEntry struct {
          e logrus.FieldLogger
      }
      
      func (l logrusEntry) WithFields(fs log.Fields) log.Entry {
          return logrusEntry{
              e: l.e.WithFields(logrus.Fields(fs)),
          }
      }
      
      func (l logrusEntry) WithField(name string, value interface{}) log.Entry {
          return logrusEntry{
              e: l.e.WithField(name, value),
          }
      }
      
      func (l logrusEntry) Debug(args ...interface{}) {
          l.e.Debug(args...)
      }
      
      func (l logrusEntry) Info(args ...interface{}) {
          l.e.Info(args...)
      }
      
      func (l logrusEntry) Warn(args ...interface{}) {
          l.e.Warn(args...)
      }
      
      func (l logrusEntry) Error(args ...interface{}) {
          l.e.Error(args...)
      }
      
      func (l logrusEntry) Debugf(format string, args ...interface{}) {
          l.e.Debugf(format, args...)
      }
      
      func (l logrusEntry) Infof(format string, args ...interface{}) {
          l.e.Infof(format, args...)
      }
      
      func (l logrusEntry) Warnf(format string, args ...interface{}) {
          l.e.Warnf(format, args...)
      }
      
      func (l logrusEntry) Errorf(format string, args ...interface{}) {
          l.e.Errorf(format, args...)
      }
      2. Specify a custom log lib when creating the client.
      client, err := pulsar.NewClient(pulsar.ClientOptions{
          URL:    "pulsar://localhost:6650",
          Logger: NewLoggerWithLogrus(log.StandardLogger(), "test.log"),
      })
      The above demo shows you how to redirect the log file of the Pulsar SDK for Go to the test.log file in the current path. You can redirect the log file to a specified location as needed.
      Contact Us

      Contact our sales team or business advisors to help your business.

      Contact Us
      Technical Support

      Open a ticket if you're looking for further assistance. Our Ticket is 7x24 avaliable.

      Submit a Ticket
      7x24 Phone Support
      Copyright © 2013-2024 Tencent Cloud. All Rights Reserved.
      Privacy PolicyLegalCookie Policy