Merge pull request #30 from changsongl/develop

changsongl · web-flow · commit 0404718875c5 · 2021-07-30T13:33:19.000+08:00
feat: encoder, compress encoder like protobuf.
diff --git a/cmd/delayqueue/main.go b/cmd/delayqueue/main.go
@@ -126,6 +126,7 @@ func run() int {
 	disp := dispatch.NewDispatch(l,
 		func() (bucket.Bucket, pool.Pool, queue.Queue, timer.Timer) {
 			cli := client.New(conf.Redis)
+
 			s := redis.NewStore(cli)
 
 			b := bucket.New(s, l, conf.DelayQueue.BucketSize, conf.DelayQueue.BucketName)
diff --git a/config/config.go b/config/config.go
@@ -160,8 +160,7 @@ func (c *Conf) Load(file string, fileType FileType) error {
 		return err
 	}
 
-	err = c.load(bts, decoder.DecodeFunc())
-	if err != nil {
+	if err = c.load(bts, decoder.DecodeFunc()); err != nil {
 		return err
 	}
 
diff --git a/config/config.yaml b/config/config.yaml
@@ -9,6 +9,7 @@ delay_queue:
   timer_fetch_interval: 1000                 # fetching job interval(ms), decrease interval may get better throughout.
   timer_fetch_delay:    0                    # fetch delay(ms), if there are still job in the bucket after the fetch,
                                              # it will delay timer_fetch_delay ms for next fetch. Default is not wait.
+  #encoder:              "json"               # encoder: json, compress(default, similar to protobuf)
 
 # redis config
 redis:
diff --git a/config/config.yaml.example b/config/config.yaml.example
@@ -9,6 +9,7 @@ delay_queue:
   timer_fetch_interval: 2000        # fetching job interval(ms), decrease interval may get better throughout.
   timer_fetch_delay:    0           # fetch delay(ms), if there are still job in the bucket after the fetch,
                                     # it will delay timer_fetch_delay ms for next fetch. Default is not wait.
+  encoder:              "json"               # encoder: json, compress(fast diy encode, similar to protobuf)
 
 # redis config
 redis:
diff --git a/go.mod b/go.mod
@@ -11,6 +11,7 @@ require (
 	github.com/go-redsync/redsync/v4 v4.0.4
 	github.com/golang/mock v1.5.0
 	github.com/gomodule/redigo v2.0.0+incompatible // indirect
+	github.com/json-iterator/go v1.1.11
 	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/prometheus/client_golang v1.9.0
 	github.com/stretchr/testify v1.7.0
diff --git a/go.sum b/go.sum
@@ -184,6 +184,8 @@ github.com/json-iterator/go v1.1.8/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/u
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.11 h1:uVUAXhF2To8cbw/3xN3pxj6kk7TYKs98NIrTqPlMWAQ=
+github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
diff --git a/job/job.go b/job/job.go
@@ -10,12 +10,12 @@ import (
 
 // Job job for delay queue
 type Job struct {
-	Topic   Topic       `json:"topic"`
-	ID      ID          `json:"id"`
-	Delay   Delay       `json:"delay"`
-	TTR     TTR         `json:"ttr"`
-	Body    Body        `json:"body"`
-	Version Version     `json:"version"`
+	Topic   Topic       `json:"topic,omitempty"`
+	ID      ID          `json:"id,omitempty"`
+	Delay   Delay       `json:"delay,omitempty"`
+	TTR     TTR         `json:"ttr,omitempty"`
+	Body    Body        `json:"body,omitempty"`
+	Version Version     `json:"version,omitempty"`
 	Mutex   lock.Locker `json:"-"`
 }
 
diff --git a/pkg/encode/compress.go b/pkg/encode/compress.go
@@ -6,6 +6,8 @@ import (
 	"github.com/changsongl/delay-queue/job"
 )
 
+// |tag|[length]|data
+
 // tag s
 const (
 	TagID uint64 = iota
@@ -32,7 +34,7 @@ func NewCompress() Encoder {
 	return &compress{}
 }
 
-// Encode compress encode
+// Encode compress encode, not using reflect
 func (c *compress) Encode(j *job.Job) ([]byte, error) {
 	buf := make([]byte, c.bufLength(j))
 	written := 0
@@ -42,8 +44,6 @@ func (c *compress) Encode(j *job.Job) ([]byte, error) {
 	if !j.TTR.IsEmpty() {
 		written += c.PutUInt64(TagTTR, uint64(j.TTR), buf[written:])
 	}
-	written += c.PutUInt64(TagVersion, j.Version.UInt64(), buf[written:])
-
 	if !j.ID.IsEmpty() {
 		written += c.PutString(TagID, string(j.ID), buf[written:])
 	}
@@ -53,6 +53,7 @@ func (c *compress) Encode(j *job.Job) ([]byte, error) {
 	if !j.Topic.IsEmpty() {
 		written += c.PutString(TagTopic, string(j.Topic), buf[written:])
 	}
+	written += c.PutUInt64(TagVersion, j.Version.UInt64(), buf[written:])
 
 	return buf[:written], nil
 }
@@ -100,7 +101,7 @@ func (c *compress) Decode(b []byte, j *job.Job) error {
 func (c *compress) bufLength(j *job.Job) int {
 	l := (TagLength+MaxUInt64Length)*5 + len(j.ID) + len(j.Topic)
 	if j.Body != "" {
-		l += TagLength + MaxUInt64Length + len(j.Topic)
+		l += TagLength + MaxUInt64Length + len(j.Body)
 	}
 	return l
 }
diff --git a/pkg/encode/compress_test.go b/pkg/encode/compress_test.go
@@ -1,34 +1,10 @@
 package encode
 
 import (
-	"fmt"
-	"github.com/changsongl/delay-queue/job"
-	"github.com/changsongl/delay-queue/pkg/lock"
-	"github.com/stretchr/testify/require"
 	"testing"
 )
 
 func TestCompressEncoder(t *testing.T) {
 	encoder := NewCompress()
-	j, err := job.New("jobTopic11", "1222", 10, 5, "", func(name string) lock.Locker {
-		return nil
-	})
-	require.NoError(t, err)
-
-	str, err := encoder.Encode(j)
-	require.NoError(t, err)
-	fmt.Println(str)
-
-	jDecode := &job.Job{}
-	err = encoder.Decode(str, jDecode)
-	fmt.Printf("%+v", jDecode)
-
-	require.NoError(t, err)
-	require.Equal(t, j.ID, jDecode.ID)
-	require.Equal(t, j.TTR, jDecode.TTR)
-	require.Equal(t, j.Delay, jDecode.Delay)
-	require.Equal(t, j.Topic, jDecode.Topic)
-	require.True(t, j.Version.Equal(jDecode.Version))
-	require.Equal(t, j.Body, jDecode.Body)
-	require.Equal(t, j.Version.String(), jDecode.Version.String())
+	runEncodeTest(t, encoder)
 }
diff --git a/pkg/encode/encode_benchmark_test.go b/pkg/encode/encode_benchmark_test.go
@@ -10,7 +10,7 @@ import (
 // BenchmarkCompressEncode    	1000000000	         0.000011 ns/op   len(32)
 func BenchmarkCompressEncode(t *testing.B) {
 	encoder := NewCompress()
-	j, err := job.New("jobTopic", "131223", 1, 1, "", func(name string) lock.Locker {
+	j, err := job.New("jobTopic", "131223", 1, 1, job.Body(longText), func(name string) lock.Locker {
 		return nil
 	})
 	require.NoError(t, err)
@@ -23,7 +23,7 @@ func BenchmarkCompressEncode(t *testing.B) {
 // BenchmarkJSONEncode    	1000000000	         0.000024 ns/op    len(92)
 func BenchmarkJSONEncode(t *testing.B) {
 	encoder := NewJSON()
-	j, err := job.New("jobTopic", "131223", 1, 1, "", func(name string) lock.Locker {
+	j, err := job.New("jobTopic", "131223", 1, 1, job.Body(longText), func(name string) lock.Locker {
 		return nil
 	})
 	require.NoError(t, err)
diff --git a/pkg/encode/encode_test.go b/pkg/encode/encode_test.go
@@ -0,0 +1,70 @@
+package encode
+
+import (
+	"github.com/changsongl/delay-queue/job"
+	"github.com/changsongl/delay-queue/pkg/lock"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+var longText = `
+The Bible is not a single book but a collection of books, whose complex development is not completely understood. The books began as songs and stories orally transmitted from generation to generation before being written down in a process that began sometime around the start of the first millennium BCE and continued for over a thousand years. The Bible was written and compiled by many people, from a variety of disparate cultures, most of whom are unknown.[18] British biblical scholar John K. Riches wrote:[19]
+[T]he biblical texts were produced over a period in which the living conditions of the writers – political, cultural, economic, and ecological – varied enormously. There are texts which reflect a nomadic existence, texts from people with an established monarchy and Temple cult, texts from exile, texts born out of fierce oppression by foreign rulers, courtly texts, texts from wandering charismatic preachers, texts from those who give themselves the airs of sophisticated Hellenistic writers. It is a time-span which encompasses the compositions of Homer, Plato, Aristotle, Thucydides, Sophocles, Caesar, Cicero, and Catullus. It is a period which sees the rise and fall of the Assyrian empire (twelfth to seventh century) and of the Persian empire (sixth to fourth century), Alexander's campaigns (336–326), the rise of Rome and its domination of the Mediterranean (fourth century to the founding of the Principate, 27 BCE), the destruction of the Jerusalem Temple (70 CE), and the extension of Roman rule to parts of Scotland (84 CE).
+Hebrew Bible from 1300. page 20, Genesis.
+Hebrew Bible from 1300. Genesis.
+Considered to be scriptures (sacred, authoritative religious texts), the books were compiled by different religious communities into various biblical canons (official collections of scriptures). The earliest compilation, containing the first five books of the Bible and called the Torah (meaning "law", "instruction", or "teaching") or Pentateuch ("five books"), was accepted as Jewish canon by the 5th century BCE. A second collection of narrative histories and prophesies, called the Nevi'im ("prophets"), was canonized in the 3rd century BCE. A third collection called the Ketuvim ("writings"), containing psalms, proverbs, and narrative histories, was canonized sometime between the 2nd century BCE and the 2nd century CE. These three collections were written mostly in Hebrew, with some parts in Aramaic, and together form the Hebrew Bible or "TaNaKh" (a portmanteau of "Torah", "Nevi'im", and "Ketuvim").[20]
+Greek-speaking Jews in Alexandria and elsewhere in the Jewish diaspora considered additional scriptures, composed between 200 BCE and 100 CE and not included in the Hebrew Bible, to be canon. These additional texts were included in a translation of the Hebrew Bible into Koine Greek (common Greek spoken by ordinary people) known as the Septuagint (meaning "the work of the seventy"), which began as a translation of the Torah made around 250 BCE and continued to develop for several centuries. The Septuagint contained all of the books of the Hebrew Bible, re-organized and with some textual differences, with the additional scriptures interspersed throughout.[21]
+Saint Paul Writing His Epistles, 16th-century painting.
+During the rise of Christianity in the 1st century CE, new scriptures were written in Greek about the life and teachings of Jesus Christ, who Christians believed was the messiah prophesized in the books of the Hebrew Bible. Two collections of these new scriptures – the Pauline epistles and the Gospels – were accepted as canon by the end of the 2nd century CE. A third collection, the catholic epistles, were canonized over the next few centuries. Christians called these new scriptures the "New Testament", and began referring to the Septuagint as the "Old Testament".[22]
+Between 385 and 405 CE, the early Christian church translated its canon into Vulgar Latin (the common Latin spoken by ordinary people), a translation known as the Vulgate, which included in its Old Testament the books that were in the Septuagint but not in the Hebrew Bible. The Vulgate introduced stability to the Bible, but also began the East-West Schism between Latin-speaking Western Christianity (led by the Catholic Church) and multi-lingual Eastern Christianity (led by the Eastern Orthodox Church). Christian denominations' biblical canons varied not only in the language of the books, but also in their selection, organization, and text.[23]
+Jewish rabbis began developing a standard Hebrew Bible in the 1st century CE, maintained since the middle of the first millennium by the Masoretes, and called the Masoretic Text. Christians have held ecumenical councils to standardize their biblical canon since the 4th century CE. The Council of Trent (1545–63), held by the Catholic Church in response to the Protestant Reformation, authorized the Vulgate as its official Latin translation of the Bible. The Church deemed the additional books in its Old Testament that were interspersed among the Hebrew Bible books to be "deuterocanonical" (meaning part of a second or later canon). Protestant Bibles either separated these books into a separate section called the "Apocrypha" (meaning "hidden away") between the Old and New Testaments, or omitted them altogether. The 17th-century Protestant King James Version was the most ubiquitous English Bible of all time, but it has largely been superseded by modern translations.[24]`
+
+func runEncodeTest(t *testing.T, encoder Encoder) {
+	for _, j := range testCases(t) {
+		str, err := encoder.Encode(j)
+		require.NoError(t, err)
+
+		jDecode := &job.Job{}
+		err = encoder.Decode(str, jDecode)
+		t.Logf("%+v", jDecode)
+
+		require.NoError(t, err)
+		require.Equal(t, j.ID, jDecode.ID)
+		require.Equal(t, j.TTR, jDecode.TTR)
+		require.Equal(t, j.Delay, jDecode.Delay)
+		require.Equal(t, j.Topic, jDecode.Topic)
+		require.True(t, j.Version.Equal(jDecode.Version))
+		require.Equal(t, j.Body, jDecode.Body)
+		require.Equal(t, j.Version.String(), jDecode.Version.String())
+	}
+}
+
+func testCases(t *testing.T) []*job.Job {
+	jEmpty, err := job.New("jobTopic", "1sdsa", 0, 0, "", func(name string) lock.Locker {
+		return nil
+	})
+	require.NoError(t, err)
+
+	jAllEmpty := &job.Job{}
+
+	jShort, err := job.New("jobTopicdsad", "1sdsadsads", 10,
+		20, "gfdgsdas", func(name string) lock.Locker {
+			return nil
+		},
+	)
+	require.NoError(t, err)
+
+	jLong, err := job.New(job.Topic(longText), job.ID(longText), 10000000000,
+		10000000000, job.Body(longText), func(name string) lock.Locker {
+			return nil
+		},
+	)
+	require.NoError(t, err)
+
+	return []*job.Job{
+		jEmpty,
+		jAllEmpty,
+		jShort,
+		jLong,
+	}
+}
diff --git a/pkg/encode/json.go b/pkg/encode/json.go
@@ -1,8 +1,8 @@
 package encode
 
 import (
-	"encoding/json"
 	"github.com/changsongl/delay-queue/job"
+	jsoniter "github.com/json-iterator/go"
 )
 
 // implemented Encoder interface
@@ -16,6 +16,7 @@ func NewJSON() Encoder {
 
 // Encode encode function
 func (j jsonEncoder) Encode(i *job.Job) ([]byte, error) {
+	var json = jsoniter.ConfigCompatibleWithStandardLibrary
 	bytes, err := json.Marshal(i)
 	if err != nil {
 		return nil, err
@@ -26,5 +27,7 @@ func (j jsonEncoder) Encode(i *job.Job) ([]byte, error) {
 
 // Decode decode function
 func (j jsonEncoder) Decode(b []byte, i *job.Job) error {
+	var json = jsoniter.ConfigCompatibleWithStandardLibrary
+
 	return json.Unmarshal(b, i)
 }
diff --git a/pkg/encode/json_test.go b/pkg/encode/json_test.go
@@ -1,33 +1,10 @@
 package encode
 
 import (
-	"fmt"
-	"github.com/changsongl/delay-queue/job"
-	"github.com/changsongl/delay-queue/pkg/lock"
-	"github.com/stretchr/testify/require"
 	"testing"
 )
 
 func TestEncoder(t *testing.T) {
 	encoder := NewJSON()
-	j, err := job.New("jobTopic", "1", 1, 1, "", func(name string) lock.Locker {
-		return nil
-	})
-	require.NoError(t, err)
-
-	str, err := encoder.Encode(j)
-	require.NoError(t, err)
-
-	jDecode := &job.Job{}
-	err = encoder.Decode(str, jDecode)
-	fmt.Println(j.Version.String())
-
-	require.NoError(t, err)
-	require.Equal(t, j.ID, jDecode.ID)
-	require.Equal(t, j.TTR, jDecode.TTR)
-	require.Equal(t, j.Delay, jDecode.Delay)
-	require.Equal(t, j.Topic, jDecode.Topic)
-	require.True(t, j.Version.Equal(jDecode.Version))
-	require.Equal(t, j.Body, jDecode.Body)
-	require.Equal(t, j.Version.String(), jDecode.Version.String())
+	runEncodeTest(t, encoder)
 }
diff --git a/store/redis/store.go b/store/redis/store.go
@@ -21,5 +21,5 @@ func (s storage) GetLock(name string) lock.Locker {
 
 // NewStore return a redis storage
 func NewStore(r redis.Redis) store.Store {
-	return &storage{rds: r, encoder: encode.NewJSON()}
+	return &storage{rds: r, encoder: encode.NewCompress()}
 }
diff --git a/test/integration/integration_test.go b/test/integration/integration_test.go
@@ -31,7 +31,7 @@ func TestDelayQueueAddAndRemove(t *testing.T) {
 	for i := 0; i < Jobs; i++ {
 		delayTime := rand.Intn(DelayTimeSeconds)
 		id := fmt.Sprintf("test-%d", i)
-		j, err := job.New(topic, id, job.DelayOption(time.Duration(delayTime)*time.Second))
+		j, err := job.New(topic, id, job.DelayOption(time.Duration(delayTime)*time.Second), job.BodyOption("a body"))
 		require.NoError(t, err)
 
 		err = AddJobRecord(key, id)
@@ -50,6 +50,7 @@ func TestDelayQueueAddAndRemove(t *testing.T) {
 		ch := c.Consume()
 		for jobMsg := range ch {
 			id := jobMsg.GetID()
+			//t.Logf("%+v", jobMsg)
 			err := DeleteJobRecord(key, id)
 			require.NoError(t, err)
 

Original file line number	Diff line number	Diff line change
`@@ -160,8 +160,7 @@ func (c *Conf) Load(file string, fileType FileType) error {`
`160`	`160`	`return err`
`161`	`161`	`}`
`162`	`162`
`163`		`- err = c.load(bts, decoder.DecodeFunc())`
`164`		`- if err != nil {`
	`163`	`+ if err = c.load(bts, decoder.DecodeFunc()); err != nil {`
`165`	`164`	`return err`
`166`	`165`	`}`
`167`	`166`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,8 @@ import (`
`6`	`6`	`"github.com/changsongl/delay-queue/job"`
`7`	`7`	`)`
`8`	`8`
	`9`	`+// \|tag\|[length]\|data`
	`10`	`+`
`9`	`11`	`// tag s`
`10`	`12`	`const (`
`11`	`13`	`TagID uint64 = iota`
`@@ -32,7 +34,7 @@ func NewCompress() Encoder {`
`32`	`34`	`return &compress{}`
`33`	`35`	`}`
`34`	`36`
`35`		`-// Encode compress encode`
	`37`	`+// Encode compress encode, not using reflect`
`36`	`38`	`func (c compress) Encode(j job.Job) ([]byte, error) {`
`37`	`39`	`buf := make([]byte, c.bufLength(j))`
`38`	`40`	`written := 0`
`@@ -42,8 +44,6 @@ func (c compress) Encode(j job.Job) ([]byte, error) {`
`42`	`44`	`if !j.TTR.IsEmpty() {`
`43`	`45`	`written += c.PutUInt64(TagTTR, uint64(j.TTR), buf[written:])`
`44`	`46`	`}`
`45`		`- written += c.PutUInt64(TagVersion, j.Version.UInt64(), buf[written:])`
`46`		`-`
`47`	`47`	`if !j.ID.IsEmpty() {`
`48`	`48`	`written += c.PutString(TagID, string(j.ID), buf[written:])`
`49`	`49`	`}`
`@@ -53,6 +53,7 @@ func (c compress) Encode(j job.Job) ([]byte, error) {`
`53`	`53`	`if !j.Topic.IsEmpty() {`
`54`	`54`	`written += c.PutString(TagTopic, string(j.Topic), buf[written:])`
`55`	`55`	`}`
	`56`	`+ written += c.PutUInt64(TagVersion, j.Version.UInt64(), buf[written:])`
`56`	`57`
`57`	`58`	`return buf[:written], nil`
`58`	`59`	`}`
`@@ -100,7 +101,7 @@ func (c compress) Decode(b []byte, j job.Job) error {`
`100`	`101`	`func (c compress) bufLength(j job.Job) int {`
`101`	`102`	`l := (TagLength+MaxUInt64Length)*5 + len(j.ID) + len(j.Topic)`
`102`	`103`	`if j.Body != "" {`
`103`		`- l += TagLength + MaxUInt64Length + len(j.Topic)`
	`104`	`+ l += TagLength + MaxUInt64Length + len(j.Body)`
`104`	`105`	`}`
`105`	`106`	`return l`
`106`	`107`	`}`