beanmsq

README

beanstalkd

二进制安装 debian: sudo aptitude install beansatlkd sudo system enable beantalkd

TODO:docker 版本安装

二进制看beanstalkd中的数据

https://github.com/davidpersson/bsa go get github.com/davidpersson/bsa

使用列子详见example目录

Documentation

Overview

例子

c := NewConsumer("localhost:11130", "test")

handle := func(ctx context.Context, job *beans.Job, tried int) bool{
		// 处理结束后返回true删除数据
		return true
}

开启两个队列去并发读取 go c.Reserve(20*time.Minute, handle) // go c.Reserve(20*time.Minute, handle) // 警告：目前beanstalkd消费者并发连接读取测试未通过, 多线程读取时注意去重操作

// 在适当的地方关闭连接 // c.Close() // Stop func to stop working

Index

• Constants
• Variables
• func IsErrNotFound(err error) bool
• type Consumer
  • func NewConsumer(addr, tube string) Consumer
• type HandleContext
• type Job
• type Producer
  • func NewProducer(size int, addr, tube string) Producer

Constants

const MAX_TRY_TIMES = 48 + 30 + 1

最大推送次数

Variables

var ErrClosed = errors.New("msq: closed")

ErrClosed closed by Close

Functions

func IsErrNotFound

func IsErrNotFound(err error) bool

Types

type Consumer

type Consumer interface {
	io.Closer

	// reserveOut -- 与服务器连接的存活检查时间
	// handle -- 接收处理函数
	Reserve(reserveOut time.Duration, handle HandleContext) error
}

func NewConsumer

func NewConsumer(addr, tube string) Consumer

type HandleContext

type HandleContext func(ctx context.Context, job *Job, tried int) bool

若发送不成功 返回true删除beanstalkd队件数据，否则不删除在一定时间后放回到就绪队中再次读取以便达到重试的效果。 重试机制分别间隔以1次3秒钟、30次每分钟、48次每小时再次尝试发送, 若48小时后未能发送成功，数据将被强制删除。 已放回就绪队列的次数通过tried进行了推送

type Job

type Job struct {
	ID   uint64
	Body []byte
}

type Producer

type Producer interface {
	io.Closer
	Put(data []byte) error
}

Producer is the interface that wrap the Put method

func NewProducer

func NewProducer(size int, addr, tube string) Producer

NewProducer create Producer object.