NSQ源碼(二)-發(fā)送消息

go-nsq發(fā)送消息流程:


config := nsq.NewConfig()
// 創(chuàng)建producer
producer, err := nsq.NewProducer("192.168.200.151:4150", config)
if err != nil {
    log.Fatal(err)
}
messageBody := []byte(msg)
topicName := "test"
// Synchronously publish a single message to the specified topic.
// Messages can also be sent asynchronously and/or in batches.
err := producer.Publish(topicName, messageBody)

發(fā)送消息的時候

func (w *Producer) Publish(topic string, body []byte) error {
    return w.sendCommand(Publish(topic, body))
}
// 構(gòu)建publish command
func Publish(topic string, body []byte) *Command {
    var params = [][]byte{[]byte(topic)}
    return &Command{[]byte("PUB"), params, body}
}
func (w *Producer) sendCommand(cmd *Command) error {
    doneChan := make(chan *ProducerTransaction)
    err := w.sendCommandAsync(cmd, doneChan, nil)
    if err != nil {
        close(doneChan)
        return err
    }
    t := <-doneChan  //同步發(fā)送世杀，等待返回
    return t.Error
}
func (w *Producer) sendCommandAsync(cmd *Command, doneChan chan *ProducerTransaction,
    args []interface{}) error {
    // keep track of how many outstanding producers we're dealing with
    // in order to later ensure that we clean them all up...
    atomic.AddInt32(&w.concurrentProducers, 1)
    defer atomic.AddInt32(&w.concurrentProducers, -1)

    if atomic.LoadInt32(&w.state) != StateConnected {
        err := w.connect()  // 注意這里connect
        if err != nil {
            return err
        }
    }

    t := &ProducerTransaction{
        cmd:      cmd,
        doneChan: doneChan,
        Args:     args,
    }

    select {
    case w.transactionChan <- t:
    case <-w.exitChan:
        return ErrStopped
    }

    return nil
}

在producer的connect中調(diào)用conn的Connect

_, err := w.conn.Connect()

在創(chuàng)建連接之后亚享，需要發(fā)送下面的數(shù)據(jù)

func (c *Conn) Connect() (*IdentifyResponse, error) {
    dialer := &net.Dialer{
        LocalAddr: c.config.LocalAddr,
        Timeout:   c.config.DialTimeout,
    }

    conn, err := dialer.Dial("tcp", c.addr)
    if err != nil {
        return nil, err
    }
    c.conn = conn.(*net.TCPConn)
    c.r = conn
    c.w = conn
        // 這里的MagicV2是 "  V2"
    _, err = c.Write(MagicV2)
    if err != nil {
        c.Close()
        return nil, fmt.Errorf("[%s] failed to write magic - %s", c.addr, err)
    }
        // 這里發(fā)送command "IDENTIFY"
    resp, err := c.identify()
    if err != nil {
        return nil, err
    }

    if resp != nil && resp.AuthRequired {
        if c.config.AuthSecret == "" {
            c.log(LogLevelError, "Auth Required")
            return nil, errors.New("Auth Required")
        }
        err := c.auth(c.config.AuthSecret)
        if err != nil {
            c.log(LogLevelError, "Auth Failed %s", err)
            return nil, err
        }
    }

    c.wg.Add(2)
    atomic.StoreInt32(&c.readLoopRunning, 1)
    go c.readLoop()
    go c.writeLoop()
    return resp, nil
}

從上面代碼中看到