gobox中的consumer处理框架

我们都会有从异步队列中消费的需求，今天来说下gobox中的consumer处理框架

consumer处理架构图

gobox-consumer.png

重要的对象

IMessage

定义每条消息

type IMessage interface {
    Body() []byte
}

ConsumerHandleFunc

consumer中从队列收到每条消息后，调用这个方法

type ConsumerHandleFunc func(message IMessage) error

IConsumer

定义消费者行为

type IConsumer interface {
    SetHandleFunc(hf ConsumerHandleFunc)
    Start()
    Stop()
}

NewWorkerFunc

每个Worker的构造方法

type NewWorkerFunc func() IWorker

IWorker

定义Worker

type IWorker interface {
    SetWorkId(id int)
    SetLogger(logger golog.ILogger)

    Work(wg *sync.WaitGroup, lineCh chan []byte, stopCh chan bool)
}

LineProcessFunc

每条消息在Worker中的实际处理方法

type LineProcessFunc func(line []byte) error

BaseWorker

框架提供的一个简单基础Worker对象，组合这个对象后，只需要实现LineProcessFunc即可

type BaseWorker struct

Task

Task用于实现consumer的处理框架

使用示例

package main

import (
    "github.com/goinbox/goconsumer"

    "fmt"
    "strconv"
    "time"
)

// 这里实现Worker
type DemoWorker struct {
    *goconsumer.BaseWorker
}

func NewDemoWorker() goconsumer.IWorker {
    worker := &DemoWorker{goconsumer.NewBaseWorker()}
    worker.SetLineProcessFunc(worker.LineProcessFunc)

    return worker
}

func (d *DemoWorker) LineProcessFunc(line []byte) error {
    idStr := strconv.Itoa(d.Id)
    fmt.Println("wid:" + idStr + " process line:" + string(line))

    return nil
}

// 这里实现Message
type DemoMessage struct {
    body []byte
}

func (d *DemoMessage) Body() []byte {
    return d.body
}

// 这里实现一个简单的Consumer，模拟从队列中获得100条消息
type DemoConsumer struct {
    hf goconsumer.ConsumerHandleFunc
}

func (d *DemoConsumer) SetHandleFunc(hf goconsumer.ConsumerHandleFunc) {
    d.hf = hf
}

func (d *DemoConsumer) Start() {
    for i := 0; i < 100; i++ {
        str := "This message is from DemoConsumer loop " + strconv.Itoa(i)
        d.hf(&DemoMessage{[]byte(str)})
    }

    time.Sleep(time.Second * 1)
}

func (d *DemoConsumer) Stop() {

}


// 执行Task任务，调用consumer处理框架
func main() {
    task := goconsumer.NewTask("Demo")
    consumer := new(DemoConsumer)

    task.SetConsumer(consumer).
        SetWorker(10, NewDemoWorker).
        Start()
}

欢迎大家使用，使用中有遇到问题随时反馈，我们会尽快响应，谢谢！