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

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

 1package main
 2
 3import (
 4	"testing"
 5	"time"
 6
 7	"github.com/apache/pulsar-client-go/pulsar"
 8)
 9
10func TestNewClient(t *testing.T) {
11	client, err := pulsar.NewClient(pulsar.ClientOptions{
12		URL:               "pulsar://192.168.2.13:6650",
13		OperationTimeout:  30 * time.Second,
14		ConnectionTimeout: 30 * time.Second,
15	})
16	if err != nil {
17		t.Fatal(err)
18	}
19	t.Log(client)
20	client.Close()
21}

 1// Client represents a pulsar client
 2type Client interface {
 3	// CreateProducer Creates the producer instance
 4	// This method will block until the producer is created successfully
 5	CreateProducer(ProducerOptions) (Producer, error)
 6
 7	// Subscribe Creates a `Consumer` by subscribing to a topic.
 8	//
 9	// If the subscription does not exist, a new subscription will be created and all messages published after the
10	// creation will be retained until acknowledged, even if the consumer is not connected
11	Subscribe(ConsumerOptions) (Consumer, error)
12
13	// CreateReader Creates a Reader instance.
14	// This method will block until the reader is created successfully.
15	CreateReader(ReaderOptions) (Reader, error)
16
17	// TopicPartitions Fetches the list of partitions for a given topic
18	//
19	// If the topic is partitioned, this will return a list of partition names.
20	// If the topic is not partitioned, the returned list will contain the topic
21	// name itself.
22	//
23	// This can be used to discover the partitions and create {@link Reader},
24	// {@link Consumer} or {@link Producer} instances directly on a particular partition.
25	TopicPartitions(topic string) ([]string, error)
26
27	// Close Closes the Client and free associated resources
28	Close()
29}

 1func TestProducer(t *testing.T) {
 2	client, err := pulsar.NewClient(pulsar.ClientOptions{
 3		URL:               "pulsar://192.168.2.13:6650",
 4		OperationTimeout:  30 * time.Second,
 5		ConnectionTimeout: 30 * time.Second,
 6	})
 7	if err != nil {
 8		t.Fatal(err)
 9	}
10	defer client.Close()
11
12	producer, err := client.CreateProducer(pulsar.ProducerOptions{
13		Name:  "TestProducer-Go",
14		Topic: "persistent://study/app1/topic-1",
15	})
16	if err != nil {
17		t.Fatal(err)
18	}
19	defer producer.Close()
20	msg := &pulsar.ProducerMessage{
21		Key:     "msgKey1",
22		Payload: []byte("hello go"),
23		Properties: map[string]string{
24			"p1": "v1",
25			"p2": "v2",
26		},
27	}
28	msgID, err := producer.Send(context.Background(), msg)
29	if err != nil {
30		t.Fatal(err)
31	}
32	t.Log(msgID)
33}

 1func TestConsumer(t *testing.T) {
 2	client, err := pulsar.NewClient(pulsar.ClientOptions{
 3		URL:               "pulsar://192.168.2.13:6650",
 4		OperationTimeout:  30 * time.Second,
 5		ConnectionTimeout: 30 * time.Second,
 6	})
 7	if err != nil {
 8		t.Fatal(err)
 9	}
10	defer client.Close()
11
12	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
13		Topic:            "persistent://study/app1/topic-1",
14		SubscriptionName: "sub-2",          // 订阅名称
15		Type:             pulsar.Exclusive, // 订阅类型: 独占模式
16	})
17	if err != nil {
18		t.Fatal(err)
19	}
20	defer consumer.Close()
21
22	for {
23		msg, err := consumer.Receive(context.Background())
24		if err != nil {
25			t.Fatal(err)
26		}
27		if err := processMsg(msg); err != nil {
28			consumer.Nack(msg)
29		} else {
30			consumer.Ack(msg)
31		}
32
33	}
34}
35
36func processMsg(msg pulsar.Message) error {
37	fmt.Printf("consume: %s \n", msg.Payload())
38	return nil
39}

 1func (c *consumer) Receive(ctx context.Context) (message Message, err error) {
 2	for {
 3		select {
 4		case <-c.closeCh:
 5			return nil, newError(ConsumerClosed, "consumer closed")
 6		case cm, ok := <-c.messageCh:
 7			if !ok {
 8				return nil, newError(ConsumerClosed, "consumer closed")
 9			}
10			return cm.Message, nil
11		case <-ctx.Done():
12			return nil, ctx.Err()
13		}
14	}
15}
16
17func (c *consumer) Chan() <-chan ConsumerMessage {
18	return c.messageCh
19}

1type ConsumerOptions struct {
2    ...
3	Topics []string
4	TopicsPattern string
5    ...
6}

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

 1reader, err := client.CreateReader(pulsar.ReaderOptions{
 2        Topic:          "topic-1",
 3        StartMessageID: pulsar.EarliestMessageID(),
 4    })
 5if err != nil {
 6    log.Fatal(err)
 7}
 8defer reader.Close()
 9
10for reader.HasNext() {
11    msg, err := reader.Next(context.Background())
12    if err != nil {
13        log.Fatal(err)
14    }
15
16    fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
17        msg.ID(), string(msg.Payload()))
18}