Goque provides embedded, disk-based implementations of stack and queue data structures.
Motivation for creating this project was the need for a persistent priority queue that remained performant while growing well beyond the available memory of a given machine. While there are many packages for Go offering queues, they all seem to be memory based and/or standalone solutions that are not embeddable within an application.
Instead of using an in-memory heap structure to store data, everything is stored using the Go port of LevelDB. This results in very little memory being used no matter the size of the database, while read and write performance remains near constant.
- Provides stack (LIFO), queue (FIFO), priority queue, and prefix queue structures.
- Stacks and queues (but not priority queues or prefix queues) are interchangeable.
- Persistent, disk-based.
- Optimized for fast inserts and reads.
- Goroutine safe.
- Designed to work with large datasets outside of RAM/memory.
Fetch the package from GitHub:
go get github.com/beeker1121/goque
Import to your project:
import "github.com/beeker1121/goque"
Stack is a LIFO (last in, first out) data structure.
Create or open a stack:
s, err := goque.OpenStack("data_dir")
...
defer s.Close()
Push an item:
item, err := s.Push([]byte("item value"))
// or
item, err := s.PushString("item value")
// or
item, err := s.PushObject(Object{X:1})
// or
item, err := s.PushObjectAsJSON(Object{X:1})
Pop an item:
item, err := s.Pop()
...
fmt.Println(item.ID) // 1
fmt.Println(item.Key) // [0 0 0 0 0 0 0 1]
fmt.Println(item.Value) // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value
// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
// Decode to object from JSON.
var obj Object
err := item.ToObjectFromJSON(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
Peek the next stack item:
item, err := s.Peek()
// or
item, err := s.PeekByOffset(1)
// or
item, err := s.PeekByID(1)
Update an item in the stack:
item, err := s.Update(1, []byte("new value"))
// or
item, err := s.UpdateString(1, "new value")
// or
item, err := s.UpdateObject(1, Object{X:2})
// or
item, err := s.UpdateObjectAsJSON(1, Object{X:2})
Delete the stack and underlying database:
s.Drop()
Queue is a FIFO (first in, first out) data structure.
Create or open a queue:
q, err := goque.OpenQueue("data_dir")
...
defer q.Close()
Enqueue an item:
item, err := q.Enqueue([]byte("item value"))
// or
item, err := q.EnqueueString("item value")
// or
item, err := q.EnqueueObject(Object{X:1})
// or
item, err := q.EnqueueObjectAsJSON(Object{X:1})
Dequeue an item:
item, err := q.Dequeue()
...
fmt.Println(item.ID) // 1
fmt.Println(item.Key) // [0 0 0 0 0 0 0 1]
fmt.Println(item.Value) // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value
// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
// Decode to object from JSON.
var obj Object
err := item.ToObjectFromJSON(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
Peek the next queue item:
item, err := q.Peek()
// or
item, err := q.PeekByOffset(1)
// or
item, err := q.PeekByID(1)
Update an item in the queue:
item, err := q.Update(1, []byte("new value"))
// or
item, err := q.UpdateString(1, "new value")
// or
item, err := q.UpdateObject(1, Object{X:2})
// or
item, err := q.UpdateObjectAsJSON(1, Object{X:2})
Delete the queue and underlying database:
q.Drop()
PriorityQueue is a FIFO (first in, first out) queue with priority levels.
Create or open a priority queue:
pq, err := goque.OpenPriorityQueue("data_dir", goque.ASC)
...
defer pq.Close()
Enqueue an item:
item, err := pq.Enqueue(0, []byte("item value"))
// or
item, err := pq.EnqueueString(0, "item value")
// or
item, err := pq.EnqueueObject(0, Object{X:1})
// or
item, err := pq.EnqueueObjectAsJSON(0, Object{X:1})
Dequeue an item:
item, err := pq.Dequeue()
// or
item, err := pq.DequeueByPriority(0)
...
fmt.Println(item.ID) // 1
fmt.Println(item.Priority) // 0
fmt.Println(item.Key) // [0 58 0 0 0 0 0 0 0 1]
fmt.Println(item.Value) // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value
// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
// Decode to object from JSON.
var obj Object
err := item.ToObjectFromJSON(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
Peek the next priority queue item:
item, err := pq.Peek()
// or
item, err := pq.PeekByOffset(1)
// or
item, err := pq.PeekByPriorityID(0, 1)
Update an item in the priority queue:
item, err := pq.Update(0, 1, []byte("new value"))
// or
item, err := pq.UpdateString(0, 1, "new value")
// or
item, err := pq.UpdateObject(0, 1, Object{X:2})
// or
item, err := pq.UpdateObjectAsJSON(0, 1, Object{X:2})
Delete the priority queue and underlying database:
pq.Drop()
PrefixQueue is a FIFO (first in, first out) data structure that separates each given prefix into its own queue.
Create or open a prefix queue:
pq, err := goque.OpenPrefixQueue("data_dir")
...
defer pq.Close()
Enqueue an item:
item, err := pq.Enqueue([]byte("prefix"), []byte("item value"))
// or
item, err := pq.EnqueueString("prefix", "item value")
// or
item, err := pq.EnqueueObject([]byte("prefix"), Object{X:1})
// or
item, err := pq.EnqueueObjectAsJSON([]byte("prefix"), Object{X:1})
Dequeue an item:
item, err := pq.Dequeue([]byte("prefix"))
// or
item, err := pq.DequeueString("prefix")
...
fmt.Println(item.ID) // 1
fmt.Println(item.Key) // [112 114 101 102 105 120 0 0 0 0 0 0 0 0 1]
fmt.Println(item.Value) // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value
// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
// Decode to object from JSON.
var obj Object
err := item.ToObjectFromJSON(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}
Peek the next prefix queue item:
item, err := pq.Peek([]byte("prefix"))
// or
item, err := pq.PeekString("prefix")
// or
item, err := pq.PeekByID([]byte("prefix"), 1)
// or
item, err := pq.PeekByIDString("prefix", 1)
Update an item in the prefix queue:
item, err := pq.Update([]byte("prefix"), 1, []byte("new value"))
// or
item, err := pq.UpdateString("prefix", 1, "new value")
// or
item, err := pq.UpdateObject([]byte("prefix"), 1, Object{X:2})
// or
item, err := pq.UpdateObjectAsJSON([]byte("prefix"), 1, Object{X:2})
Delete the prefix queue and underlying database:
pq.Drop()
Benchmarks were ran on a Google Compute Engine n1-standard-1 machine (1 vCPU 3.75 GB of RAM):
Go 1.6:
$ go test -bench=.
PASS
BenchmarkPriorityQueueEnqueue 200000 8104 ns/op 522 B/op 7 allocs/op
BenchmarkPriorityQueueDequeue 200000 18622 ns/op 1166 B/op 17 allocs/op
BenchmarkQueueEnqueue 200000 8049 ns/op 487 B/op 7 allocs/op
BenchmarkQueueDequeue 200000 18970 ns/op 1089 B/op 17 allocs/op
BenchmarkStackPush 200000 8145 ns/op 487 B/op 7 allocs/op
BenchmarkStackPop 200000 18947 ns/op 1097 B/op 17 allocs/op
ok github.com/beeker1121/goque 22.549s
Go 1.8:
$ go test -bench=.
BenchmarkPrefixQueueEnqueue 20000 60553 ns/op 10532 B/op 242 allocs/op
BenchmarkPrefixQueueDequeue 10000 100727 ns/op 18519 B/op 444 allocs/op
BenchmarkPriorityQueueEnqueue 300000 4781 ns/op 557 B/op 9 allocs/op
BenchmarkPriorityQueueDequeue 200000 11656 ns/op 1206 B/op 19 allocs/op
BenchmarkQueueEnqueue 300000 4625 ns/op 513 B/op 9 allocs/op
BenchmarkQueueDequeue 200000 11537 ns/op 1125 B/op 19 allocs/op
BenchmarkStackPush 300000 4631 ns/op 513 B/op 9 allocs/op
BenchmarkStackPop 200000 9629 ns/op 1116 B/op 19 allocs/op
PASS
ok github.com/beeker1121/goque 18.135s
syndtr (https://github.com/syndtr) - LevelDB port to Go
bogdanovich (https://github.com/bogdanovich/siberite) - Server based queue for Go using LevelDB
connor4312 (https://github.com/connor4312) - Recommending BoltDB/LevelDB, helping with structure
bwmarrin (https://github.com/bwmarrin) - Recommending BoltDB/LevelDB
zeroZshadow (https://github.com/zeroZshadow) - Code review and optimization
nstafie (https://github.com/nstafie) - Help with structure