golang异步kafka生产者

在实际业务场景中，为了提高系统的实时性，减轻日志存储压力，需要将日志直接生产至消息中间件，减少flume或flumted收集所导致的延时及性能压力，本文实现了一下功能：
实现了一个静态调用的异步生产者    AsyncProducer
封装了一个用于异步发送的生产器    Agent

//@description	kafka代理
//@author chenbintao
//@data	2017-09-27	10:30	初稿
//		2017-09-27	11:15	规范代码
//		2017-09-28	14:15	对发送逻辑进行了优化

package kafkaAgent

import (
    "fmt"
    "log"
    "runtime/debug"
    "strings"
    "time"

    "github.com/Shopify/sarama"
)

const (
    _BROKER_LIST_ = `localhost:9092`
)
const (
    _LABEL_ = "[_kafkaAgent_]"
)

var (
    IS_DEBUG = false
    _PAUSE_  = false
)

func SetDebug(debug bool) {
    IS_DEBUG = debug
}

type Agent struct {
    flag           bool
    BrokerList     string
    TopicList      string
    SendTimeOut    time.Duration
    ReceiveTimeOut time.Duration
    AsyncProducer  sarama.AsyncProducer
}

func (this *Agent) Set(BrokerList, TopicList string, SendTimeOut, ReceiveTimeOut time.Duration) bool {
    //只允许初始化一次
    if this.flag {

        return false
    }

    this.flag = true
    this.BrokerList = BrokerList
    this.TopicList = TopicList
    this.SendTimeOut = SendTimeOut
    this.ReceiveTimeOut = ReceiveTimeOut
    this.AsyncProducer = getProducer(this.BrokerList, this.SendTimeOut, true)
    if nil == this.AsyncProducer {

        return false
    }

    return this.Check()
}
func (this *Agent) Check() bool {
    if "" == this.BrokerList || "" == this.TopicList {

        return false
    }
    if 0 == this.SendTimeOut && 0 == this.ReceiveTimeOut {

        return false
    }

    return true
}
func (this *Agent) Send(msg string) bool {
    defer func() {
        if e, ok := recover().(error); ok {
            log.Println(_LABEL_, "WARN: panic in %v", e)
            log.Println(_LABEL_, string(debug.Stack()))
            this.AsyncProducer.Close()
            this.AsyncProducer = getProducer(this.BrokerList, this.SendTimeOut, true)
        }
    }()
    if !this.Check() {

        return false
    }

    return asyncProducer(
        this.AsyncProducer,
        this.TopicList,
        msg,
    )
}

//=========================================================================
// asyncProducer 异步生产者
func AsyncProducer(kafka_list, topics, s string, timeout time.Duration) bool {
    if "" == kafka_list || "" == topics {

        return false
    }
    producer := getProducer(kafka_list, timeout, false)
    if nil == producer {

        return false
    }
    defer producer.Close()
    go func(p sarama.AsyncProducer) {
        errors := p.Errors()
        success := p.Successes()
        for {
            select {
            case err := <-errors:
                if err != nil {
                    if IS_DEBUG {
                        log.Println(_LABEL_, err)
                    }
                    return
                } else {
                    return
                }
            case <-success:
                return
            }
        }
    }(producer)

    return asyncProducer(producer, topics, s)
}
func asyncProducer(p sarama.AsyncProducer, topics, s string) bool {
    if nil == p {

        return false
    }

    msg := &sarama.ProducerMessage{
        Topic: topics,
        Value: sarama.ByteEncoder(s),
    }
    p.Input() <- msg

    if IS_DEBUG {
        fmt.Println(_LABEL_, msg)
    }

    return true
}
func getProducer(kafka_list string, timeout time.Duration, monitor bool) sarama.AsyncProducer {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    config.Producer.Timeout = timeout
    producer, err := sarama.NewAsyncProducer(strings.Split(kafka_list, ","), config)
    if err != nil {
        if IS_DEBUG {
            log.Println(_LABEL_, err)
        }
    }
    if monitor {
        //消费状态消息,防止死锁
        go func(producer sarama.AsyncProducer) {
            if nil == producer {
                log.Println(_LABEL_, "getProducer() producer error!")

                return
            }
            errors := producer.Errors()
            success := producer.Successes()
            for {
                select {
                case err := <-errors:
                    if err != nil {
                        if IS_DEBUG {
                            log.Println(_LABEL_, err)
                        }
                        continue
                    } else {
                        continue
                    }
                case <-success:
                    continue
                }
            }
        }(producer)
    }

    return producer
}