Pulsar Go client

您可以使用 Pulsar Go 客户端来创建使用 Go 语言的 Pulsar 、消费者（consumer）和。

API docs available as well
For standard API docs, consult the Godoc.

You can install the library locally using go get.

Once installed locally, you can import it into your project:

import "github.com/apache/pulsar-client-go/pulsar"

连接 URL

To connect to Pulsar using client libraries, you need to specify a Pulsar protocol URL.

Pulsar protocol URLs are assigned to specific clusters, use the pulsar scheme and have a default port of 6650. Here’s an example for localhost:

pulsar://localhost:6650

如果你有多个brokers，你可以使用下面的方法设置URl

pulsar://localhost:6550,localhost:6651,localhost:6652

A URL for a production Pulsar cluster may look something like this:

pulsar://pulsar.us-west.example.com:6650

If you’re using authentication, the URL will look like something like this:

pulsar+ssl://pulsar.us-west.example.com:6651

创建客户端

import (
    "log"
    "time"
    "github.com/apache/pulsar-client-go/pulsar"
)
func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL:               "pulsar://localhost:6650",
        OperationTimeout:  30 * time.Second,
        ConnectionTimeout: 30 * time.Second,
    })
    if err != nil {
        log.Fatalf("Could not instantiate Pulsar client: %v", err)
    }
    defer client.Close()
}

如果你有多个broker，你可以想这样初始化PulsarClient.

import (
    "log"
    "time"
    "github.com/apache/pulsar-client-go/pulsar"
)
func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL: "pulsar://localhost:6650,localhost:6651,localhost:6652",
        OperationTimeout:  30 * time.Second,
        ConnectionTimeout: 30 * time.Second,
    })
    if err != nil {
        log.Fatalf("Could not instantiate Pulsar client: %v", err)
    }
    defer client.Close()
}

The following configurable parameters are available for Pulsar clients:

Name | Description | Default | :———— | :————— |:————— | | URL | Configure the service URL for the Pulsar service.

If you have multiple brokers, you can set multiple Pulsar cluster addresses for a client.

Pulsar producers publish messages to Pulsar topics. You can Go producers using a ProducerOptions object. 下面是一个示例：

producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic: "my-topic",
})
if err != nil {
    log.Fatal(err)
}
_, err = producer.Send(context.Background(), &pulsar.ProducerMessage{
    Payload: []byte("hello"),
})
defer producer.Close()
if err != nil {
    fmt.Println("Failed to publish message", err)
}
fmt.Println("Published message")

Producer operations

Pulsar Go producers have the following methods available:

生产者示例

如何在生产者中使用消息路由器

How to use schema interface in producer

type testJSON struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
}

var (
    exampleSchemaDef = "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\"," +
        "\"fields\":[{\"name\":\"ID\",\"type\":\"int\"},{\"name\":\"Name\",\"type\":\"string\"}]}"
)

client, err := NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
if err != nil {
    log.Fatal(err)
}
defer client.Close()
properties := make(map[string]string)
properties["pulsar"] = "hello"
jsonSchemaWithProperties := NewJSONSchema(exampleSchemaDef, properties)
producer, err := client.CreateProducer(ProducerOptions{
    Topic:  "jsonTopic",
    Schema: jsonSchemaWithProperties,
})
assert.Nil(t, err)
_, err = producer.Send(context.Background(), &ProducerMessage{
    Value: &testJSON{
        ID:   100,
        Name: "pulsar",
    },
})
if err != nil {
    log.Fatal(err)
}
producer.Close()

如何在生产者中使用相应的延迟

client, err := NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
if err != nil {
    log.Fatal(err)
}
defer client.Close()
topicName := newTopicName()
producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic: topicName,
})
if err != nil {
    log.Fatal(err)
}
defer producer.Close()
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:            topicName,
    SubscriptionName: "subName",
    Type:             Shared,
})
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()
ID, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
    Payload:      []byte(fmt.Sprintf("test")),
})
if err != nil {
    log.Fatal(err)
}
fmt.Println(ID)
ctx, canc := context.WithTimeout(context.Background(), 1*time.Second)
msg, err := consumer.Receive(ctx)
if err != nil {
    log.Fatal(err)
}
fmt.Println(msg.Payload())
canc()
ctx, canc = context.WithTimeout(context.Background(), 5*time.Second)
msg, err = consumer.Receive(ctx)
if err != nil {
    log.Fatal(err)
}
fmt.Println(msg.Payload())
canc()

Consumers

Pulsar consumers subscribe to one or more Pulsar topics and listen for incoming messages produced on that topic/those topics. You can configure Go consumers using a ConsumerOptions object. Here’s a basic example that uses channels:

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    SubscriptionName: "my-sub",
    Type:             pulsar.Shared,
})
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()
for i := 0; i < 10; i++ {
    msg, err := consumer.Receive(context.Background())
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
        msg.ID(), string(msg.Payload()))
    consumer.Ack(msg)
}
if err := consumer.Unsubscribe(); err != nil {
    log.Fatal(err)
}

消费者操作

Receive example

How to use regex consumer

client, err := pulsar.NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
defer client.Close()
p, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic:           topicInRegex,
    DisableBatching: true,
})
if err != nil {
    log.Fatal(err)
}
defer p.Close()
topicsPattern := fmt.Sprintf("persistent://%s/foo.*", namespace)
opts := pulsar.ConsumerOptions{
    TopicsPattern:    topicsPattern,
    SubscriptionName: "regex-sub",
}
consumer, err := client.Subscribe(opts)
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

如何使用多topic 的Consumer

func newTopicName() string {
    return fmt.Sprintf("my-topic-%v", time.Now().Nanosecond())
}
topic1 := "topic-1"
topic2 := "topic-2"
client, err := NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
if err != nil {
    log.Fatal(err)
}
topics := []string{topic1, topic2}
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topics:           topics,
    SubscriptionName: "multi-topic-sub",
})
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

如何使用消费监听器

import (
    "fmt"
    "log"
    "github.com/apache/pulsar-client-go/pulsar"
)
func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()
    channel := make(chan pulsar.ConsumerMessage, 100)
    options := pulsar.ConsumerOptions{
        Topic:            "topic-1",
        SubscriptionName: "my-subscription",
        Type:             pulsar.Shared,
    }
    options.MessageChannel = channel
    consumer, err := client.Subscribe(options)
    if err != nil {
        log.Fatal(err)
    }
    defer consumer.Close()
    // Receive messages from channel. The channel returns a struct which contains message and the consumer from where
    // the message was received. It's not necessary here since we have 1 single consumer, but the channel could be
    // shared across multiple consumers as well
    for cm := range channel {
        msg := cm.Message
        fmt.Printf("Received message  msgId: %v -- content: '%s'\n",
            msg.ID(), string(msg.Payload()))
        consumer.Ack(msg)
    }
}

如何使用消费者接收超时器

How to use schema in consumer

type testJSON struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
}

var (
    exampleSchemaDef = "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\"," +
        "\"fields\":[{\"name\":\"ID\",\"type\":\"int\"},{\"name\":\"Name\",\"type\":\"string\"}]}"
)

client, err := NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
if err != nil {
    log.Fatal(err)
}
defer client.Close()
consumerJS := NewJSONSchema(exampleSchemaDef, nil)
consumer, err := client.Subscribe(ConsumerOptions{
    Topic:                       "jsonTopic",
    SubscriptionName:            "sub-1",
    Schema:                      consumerJS,
    SubscriptionInitialPosition: SubscriptionPositionEarliest,
})
assert.Nil(t, err)
msg, err := consumer.Receive(context.Background())
assert.Nil(t, err)
err = msg.GetSchemaValue(&s)
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

Reader

Pulsar readers process messages from Pulsar topics. Readers are different from consumers because with readers you need to explicitly specify which message in the stream you want to begin with (consumers, on the other hand, automatically begin with the most recent unacked message). You can Go readers using a ReaderOptions object. 下面是一个示例：

reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:          "topic-1",
    StartMessageID: pulsar.EarliestMessageID(),
})
if err != nil {
    log.Fatal(err)
}
defer reader.Close()

Reader operations

Pulsar Go readers have the following methods available:

Reader 示例

如何使用阅读器读取“下一个”消息

Here’s an example usage of a Go reader that uses the Next() method to process incoming messages:

import (
    "context"
    "fmt"
    "log"
    "github.com/apache/pulsar-client-go/pulsar"
)
func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()
    reader, err := client.CreateReader(pulsar.ReaderOptions{
        Topic:          "topic-1",
        StartMessageID: pulsar.EarliestMessageID(),
    })
    if err != nil {
        log.Fatal(err)
    }
    defer reader.Close()
    for reader.HasNext() {
        msg, err := reader.Next(context.Background())
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
            msg.ID(), string(msg.Payload()))
    }
}

In the example above, the reader begins reading from the earliest available message (specified by pulsar.EarliestMessage). The reader can also begin reading from the latest message (pulsar.LatestMessage) or some other message ID specified by bytes using the DeserializeMessageID function, which takes a byte array and returns a MessageID object. 下面是一个示例：

lastSavedId := // Read last saved message id from external store as byte[]
reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:          "my-golang-topic",
    StartMessageID: pulsar.DeserializeMessageID(lastSavedId),
})

如何使用阅读器读取特定消息

client, err := NewClient(pulsar.ClientOptions{
    URL: lookupURL,
})
if err != nil {
    log.Fatal(err)
}
defer client.Close()
topic := "topic-1"
ctx := context.Background()
// create producer
producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic:           topic,
    DisableBatching: true,
})
if err != nil {
    log.Fatal(err)
}
defer producer.Close()
// send 10 messages
msgIDs := [10]MessageID{}
for i := 0; i < 10; i++ {
    msgID, err := producer.Send(ctx, &pulsar.ProducerMessage{
        Payload: []byte(fmt.Sprintf("hello-%d", i)),
    })
    assert.NoError(t, err)
    assert.NotNil(t, msgID)
    msgIDs[i] = msgID
}
// create reader on 5th message (not included)
reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:          topic,
    StartMessageID: msgIDs[4],
})
if err != nil {
    log.Fatal(err)
}
defer reader.Close()
// receive the remaining 5 messages
for i := 5; i < 10; i++ {
    msg, err := reader.Next(context.Background())
    if err != nil {
    log.Fatal(err)
}
// create reader on 5th message (included)
readerInclusive, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:                   topic,
    StartMessageID:          msgIDs[4],
    StartMessageIDInclusive: true,
})
if err != nil {
    log.Fatal(err)
}
defer readerInclusive.Close()

The Pulsar Go client provides a ProducerMessage interface that you can use to construct messages to producer on Pulsar topics. Here’s an example message:

msg := pulsar.ProducerMessage{
    Payload: []byte("Here is some message data"),
    Key: "message-key",
    Properties: map[string]string{
        "foo": "bar",
    },
    EventTime: time.Now(),
    ReplicationClusters: []string{"cluster1", "cluster3"},
}
if _, err := producer.send(msg); err != nil {
    log.Fatalf("Could not publish message due to: %v", err)
}

The following methods parameters are available for ProducerMessage objects:

TLS 加密和身份验证

In order to use , you’ll need to configure your client to do so:

使用 pulsar+ssl URL 类型
设置 TLSTrustCertCertsFilePath 到你的客户端和 Pulsar broker 使用的 TLS 证书路径
配置 认证 选项

OAuth2 身份验证

To use , you’ll need to configure your client to perform the following operations. This example shows how to configure OAuth2 authentication.

oauth := pulsar.NewAuthenticationOAuth2(map[string]string{
        "type":       "client_credentials",
        "issuerUrl":  "https://dev-kt-aa9ne.us.auth0.com",
        "audience":   "https://dev-kt-aa9ne.us.auth0.com/api/v2/",
        "privateKey": "/path/to/privateKey",
        "clientId":   "0Xx...Yyxeny",
    })
client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL:              "pulsar://my-cluster:6650",
})