queue

README

queue

A task queue for mitigating server pressure in high concurrency situations and improving task processing.

Get

go get -u -v github.com/LyricTian/queue

Usage

package main

import (
	"fmt"

	"github.com/LyricTian/queue"
)

func main() {
	q := queue.NewQueue(1, 10)
	q.Run()

	defer q.Terminate()

	sjob := queue.NewSyncJob("hello", func(v interface{}) (interface{}, error) {
		return fmt.Sprintf("%s,world", v), nil
	})
	q.Push(sjob)

	result := <-sjob.Wait()
	if err := sjob.Error(); err != nil {
		panic(err)
	}

	fmt.Println(result)
	// output: hello,world
}

MIT License

    Copyright (c) 2017 Lyric

Documentation

Index

• func Push(job Jober)
• func Run(maxQueues, maxWorkers int)
• func Terminate()
• type Jober
  • func NewJob(v interface{}, fn func(interface{})) Jober
• type Queue
  • func NewQueue(maxQueues, maxWorkers int) *Queue
  • func (q *Queue) Push(job Jober)
  • func (q *Queue) Run()
  • func (q *Queue) Terminate()
• type SyncJober
  • func NewSyncJob(v interface{}, fn func(interface{}) (interface{}, error)) SyncJober
• type Worker
  • func NewWorker(pool chan<- chan Jober, done func()) Worker

Examples

• Queue

Functions

func Push

func Push(job Jober)

Push put the executable task into the queue

func Run

func Run(maxQueues, maxWorkers int)

Run start running queues, specify the number of buffers, and the number of worker threads

func Terminate

func Terminate()

Terminate terminate the queue to receive the task and release the resource

Types

type Jober

type Jober interface {
	Job()
}

Jober an asynchronous task that can be executed

func NewJob

func NewJob(v interface{}, fn func(interface{})) Jober

NewJob create an asynchronous task

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Queue a task queue for mitigating server pressure in high concurrency situations and improving task processing

Example

q := NewQueue(1, 10)
q.Run()

var count int64

for i := 0; i < 10; i++ {
	job := NewJob("foo", func(v interface{}) {
		atomic.AddInt64(&count, 1)
	})
	q.Push(job)
}

q.Terminate()
fmt.Println(count)

Output:

10

func NewQueue

func NewQueue(maxQueues, maxWorkers int) *Queue

NewQueue create a queue that specifies the number of buffers and the number of worker threads

func (*Queue) Push

func (q *Queue) Push(job Jober)

Push put the executable task into the queue

func (*Queue) Run

func (q *Queue) Run()

Run start running queues

func (*Queue) Terminate

func (q *Queue) Terminate()

Terminate terminate the queue to receive the task and release the resource

type SyncJober

type SyncJober interface {
	Jober
	Wait() <-chan interface{}
	Error() error
}

SyncJober a synchronization task that can be executed

func NewSyncJob

func NewSyncJob(v interface{}, fn func(interface{}) (interface{}, error)) SyncJober

NewSyncJob create a synchronization task

type Worker

type Worker interface {
	Start()
	Terminate()
}

Worker a worker who performs a task

func NewWorker

func NewWorker(pool chan<- chan Jober, done func()) Worker

NewWorker Create a worker who performs a task, specify a work pool and a callback after the task completes