消息隊列由生產者和消費者共同協作，生產者產生消息放入隊列中，消費者從隊列中取出消息。

1.消費者
type Producer struct {
    id     int64
    addr   string
    conn   producerConn
    config Config

    logger   logger
    logLvl   LogLevel
    logGuard sync.RWMutex

    responseChan chan []byte
    errorChan    chan []byte
    closeChan    chan 
 
int

    transactionChan chan *ProducerTransaction
    transactions    []*ProducerTransaction
    state           int32

    concurrentProducers int32
    stopFlag            int32
    exitChan            chan int
    wg                  sync.WaitGroup
    guard               sync.Mutex
}

func TestProducerConnection(t 
 
*testing.T) {
    config := NewConfig()
    laddr := "127.0.0.1"

    config.LocalAddr, _ = net.ResolveTCPAddr("tcp", laddr+":0")

    w, _ := NewProducer("127.0.0.1:4150", config)
    w.SetLogger(nullLogger, LogLevelInfo)

    err := w.Publish("write_test", []byte("test"))
    if err != nil {
        t.Fatalf(
 
"should lazily connect - %s", err)
    }

    w.Stop()

    err = w.Publish("write_test", []byte("fail test"))
    if err != ErrStopped {
        t.Fatalf("should not be able to write after Stop()")
    }
}

golang-nsq消息隊列應用