Apache Pulsar学习笔记07: Pulsar的Go客户端库

1

require github.com/apache/pulsar-client-go v0.6.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

package main

import (
	"testing"
	"time"

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

func TestNewClient(t *testing.T) {
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               "pulsar://192.168.2.13:6650",
		OperationTimeout:  30 * time.Second,
		ConnectionTimeout: 30 * time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Log(client)
	client.Close()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// Client represents a pulsar client
type Client interface {
	// CreateProducer Creates the producer instance
	// This method will block until the producer is created successfully
	CreateProducer(ProducerOptions) (Producer, error)

	// Subscribe Creates a `Consumer` by subscribing to a topic.
	//
	// If the subscription does not exist, a new subscription will be created and all messages published after the
	// creation will be retained until acknowledged, even if the consumer is not connected
	Subscribe(ConsumerOptions) (Consumer, error)

	// CreateReader Creates a Reader instance.
	// This method will block until the reader is created successfully.
	CreateReader(ReaderOptions) (Reader, error)

	// TopicPartitions Fetches the list of partitions for a given topic
	//
	// If the topic is partitioned, this will return a list of partition names.
	// If the topic is not partitioned, the returned list will contain the topic
	// name itself.
	//
	// This can be used to discover the partitions and create {@link Reader},
	// {@link Consumer} or {@link Producer} instances directly on a particular partition.
	TopicPartitions(topic string) ([]string, error)

	// Close Closes the Client and free associated resources
	Close()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

func TestProducer(t *testing.T) {
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               "pulsar://192.168.2.13:6650",
		OperationTimeout:  30 * time.Second,
		ConnectionTimeout: 30 * time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer client.Close()

	producer, err := client.CreateProducer(pulsar.ProducerOptions{
		Name:  "TestProducer-Go",
		Topic: "persistent://study/app1/topic-1",
	})
	if err != nil {
		t.Fatal(err)
	}
	defer producer.Close()
	msg := &pulsar.ProducerMessage{
		Key:     "msgKey1",
		Payload: []byte("hello go"),
		Properties: map[string]string{
			"p1": "v1",
			"p2": "v2",
		},
	}
	msgID, err := producer.Send(context.Background(), msg)
	if err != nil {
		t.Fatal(err)
	}
	t.Log(msgID)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

func TestConsumer(t *testing.T) {
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               "pulsar://192.168.2.13:6650",
		OperationTimeout:  30 * time.Second,
		ConnectionTimeout: 30 * time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer client.Close()

	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            "persistent://study/app1/topic-1",
		SubscriptionName: "sub-2",          // 订阅名称
		Type:             pulsar.Exclusive, // 订阅类型: 独占模式
	})
	if err != nil {
		t.Fatal(err)
	}
	defer consumer.Close()

	for {
		msg, err := consumer.Receive(context.Background())
		if err != nil {
			t.Fatal(err)
		}
		if err := processMsg(msg); err != nil {
			consumer.Nack(msg)
		} else {
			consumer.Ack(msg)
		}

	}
}

func processMsg(msg pulsar.Message) error {
	fmt.Printf("consume: %s \n", msg.Payload())
	return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

func (c *consumer) Receive(ctx context.Context) (message Message, err error) {
	for {
		select {
		case <-c.closeCh:
			return nil, newError(ConsumerClosed, "consumer closed")
		case cm, ok := <-c.messageCh:
			if !ok {
				return nil, newError(ConsumerClosed, "consumer closed")
			}
			return cm.Message, nil
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
}

func (c *consumer) Chan() <-chan ConsumerMessage {
	return c.messageCh
}

1
2
3
4
5
6

type ConsumerOptions struct {
    ...
	Topics []string
	TopicsPattern string
    ...
}

1
2
3
4
5
6
7
8
9

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            "persistent://study/app1/topic-1",
		SubscriptionName: "sub-2",
		Type:             pulsar.Exclusive, 
		DLQ: &pulsar.DLQPolicy{
			MaxDeliveries:   10, // 最大重发次数
			DeadLetterTopic: "persistent://study/app1/dlt-1", // 死信Topic
		},
	})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

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()))
}