add pulsar

example/pq/consumer/config.yaml

@@ -0,0 +1,7 @@
+Name: pq
+Brokers:
+  - 127.0.0.1:6650
+Topic: pq
+Conns: 2
+Processors: 2
+SubscriptionName: pq

example/pq/consumer/queue.go

@@ -0,0 +1,19 @@
+package main
+
+import (
+	"fmt"
+	"github.com/tal-tech/go-queue/pq"
+	"github.com/tal-tech/go-zero/core/conf"
+)
+
+func main() {
+	var c pq.PqConf
+	conf.MustLoad("config.yaml", &c)
+
+	q := pq.MustNewQueue(c, pq.WithHandle(func(k string, v []byte, properties map[string]string) error {
+		fmt.Printf("%s => %s; %v\n", k, v, properties)
+		return nil
+	}))
+	defer q.Stop()
+	q.Start()
+}

example/pq/producer/producer.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"math/rand"
+	"strconv"
+	"time"
+
+	"github.com/tal-tech/go-queue/pq"
+	"github.com/tal-tech/go-zero/core/cmdline"
+)
+
+type message struct {
+	Key     string `json:"key"`
+	Value   string `json:"value"`
+	Payload string `json:"message"`
+}
+
+func main() {
+	pusher := pq.NewPusher([]string{
+        "127.0.0.1:6650",
+	}, "pq")
+
+	ticker := time.NewTicker(time.Millisecond)
+
+	for round := 0; round < 30; round++ {
+		select {
+		case <-ticker.C:
+			key := strconv.FormatInt(time.Now().UnixNano(), 10)
+			count := rand.Intn(100)
+			m := message{
+				Key:     key,
+				Value:   fmt.Sprintf("%d,%d", round, count),
+				Payload: fmt.Sprintf("%d,%d", round, count),
+			}
+			body, err := json.Marshal(m)
+			if err != nil {
+				log.Fatal(err)
+			}
+
+			fmt.Println(string(body))
+			if err := pusher.Push(key, body, nil); err != nil {
+				log.Fatal(err)
+			}
+		}
+	}
+
+	cmdline.EnterToContinue()
+}

go.mod

@@ -3,6 +3,7 @@ module github.com/tal-tech/go-queue
 go 1.14
 
 require (
+	github.com/apache/pulsar-client-go v0.6.0
 	github.com/beanstalkd/go-beanstalk v0.1.0
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/klauspost/compress v1.13.1 // indirect

pq/config.go

@@ -0,0 +1,12 @@
+package pq
+
+import "github.com/tal-tech/go-zero/core/service"
+
+type PqConf struct {
+	service.ServiceConf
+	Brokers          []string
+	Topic            string
+	SubscriptionName string
+	Conns            int `json:",default=1"`
+	Processors       int `json:",default=8"`
+}

pq/pusher.go

@@ -0,0 +1,110 @@
+package pq
+
+import (
+	"context"
+	"github.com/tal-tech/go-zero/core/logx"
+	"strings"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+	"github.com/tal-tech/go-zero/core/executors"
+)
+
+type (
+	PushOption func(options *chunkOptions)
+
+	Pusher struct {
+		producer pulsar.Producer
+		topic    string
+		executor *executors.ChunkExecutor
+	}
+
+	chunkOptions struct {
+		chunkSize     int
+		flushInterval time.Duration
+	}
+)
+
+func NewPusher(addrs []string, topic string, opts ...PushOption) *Pusher {
+	url := strings.Join(addrs, ",")
+	client, err := pulsar.NewClient(pulsar.ClientOptions{
+		URL:               "pulsar://" + url,
+		ConnectionTimeout: 5 * time.Second,
+		OperationTimeout:  5 * time.Second,
+	})
+	if err != nil {
+		logx.Errorf("Could not instantiate Pulsar client: %v", err)
+	}
+	producer, err := client.CreateProducer(pulsar.ProducerOptions{
+		Topic: topic,
+	})
+	if err != nil {
+		logx.Error(err)
+	}
+
+	pusher := &Pusher{
+		producer: producer,
+		topic:    topic,
+	}
+
+	pusher.executor = executors.NewChunkExecutor(func(tasks []interface{}) {
+		for i := range tasks {
+			message := tasks[i].(pulsar.ProducerMessage)
+			_, err = pusher.producer.Send(context.Background(), &message)
+		}
+
+	}, newOptions(opts)...)
+
+	return pusher
+}
+
+func (p *Pusher) Close() {
+	p.producer.Close()
+}
+
+func (p *Pusher) Name() string {
+	return p.topic
+}
+
+func (p *Pusher) Push(key string, val []byte, properties map[string]string) error {
+	msg := pulsar.ProducerMessage{
+		Key:        key,
+		Payload:    val,
+		Properties: properties,
+	}
+	if p.executor != nil {
+		// TODO
+		return p.executor.Add(msg, len(val))
+	} else {
+		_, err := p.producer.Send(context.Background(), &msg)
+		return err
+	}
+}
+
+func WithChunkSize(chunkSize int) PushOption {
+	return func(options *chunkOptions) {
+		options.chunkSize = chunkSize
+	}
+}
+
+func WithFlushInterval(interval time.Duration) PushOption {
+	return func(options *chunkOptions) {
+		options.flushInterval = interval
+	}
+}
+
+func newOptions(opts []PushOption) []executors.ChunkOption {
+	var options chunkOptions
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	var chunkOpts []executors.ChunkOption
+	if options.chunkSize > 0 {
+		chunkOpts = append(chunkOpts, executors.WithChunkBytes(options.chunkSize))
+	}
+	if options.flushInterval > 0 {
+		chunkOpts = append(chunkOpts, executors.WithFlushInterval(options.flushInterval))
+	}
+	return chunkOpts
+}