nsqd 源码,写入数据
博客专区 > 昏鸦 的博客 > 博客详情
nsqd 源码,写入数据
昏鸦 发表于4年前
nsqd 源码,写入数据
  • 发表于 4年前
  • 阅读 117
  • 收藏 1
  • 点赞 0
  • 评论 0

切记: chanel 有有自己的持久化 队列

topic 也有自己的持久化队列,两个是相互独立

给一个topic put 数据的:

// PutMessage writes to the appropriate incoming message channel

func (t *Topic) PutMessage(msg *nsq.Message) error {

    t.RLock()

    defer t.RUnlock()

    if atomic.LoadInt32(&t.exitFlag) == 1 {

        return errors.New("exiting")

    }

    t.incomingMsgChan <- msg

    atomic.AddUint64(&t.messageCount, 1)

    return nil

}


incomingMsgChan  是:incomingMsgChan:   make(chan *nsq.Message, 1),
        memoryMsgChan:     make(chan *nsq.Message, context.nsqd.options.MemQueueSize),
        
 初始化topic 会执行:
 
 t.waitGroup.Wrap(func() { t.router() })
 t.waitGroup.Wrap(func() { t.messagePump() })
 
 func (t *Topic) router() {
    var msgBuf bytes.Buffer
    for msg := range t.incomingMsgChan { //当有数据的时候执行
        select {
        case t.memoryMsgChan <- msg://memoryMsgChan 可以写入
        default:
            err := WriteMessageToBackend(&msgBuf, msg, t.backend)// 默认持久化到硬盘
            if err != nil {
                log.Printf("ERROR: failed to write message to backend - %s", err.Error())
                // theres not really much we can do at this point, you're certainly
                // going to lose messages...
            }
        }
    }

    log.Printf("TOPIC(%s): closing ... router", t.name)
}
 
 WriteMessageToBackend  会调用:
 func (d *DiskQueue) Put(data []byte) error {
    d.RLock()
    defer d.RUnlock()

    if d.exitFlag == 1 {
        return errors.New("exiting")
    }

    d.writeChan <- data 
    return <-d.writeResponseChan
}
 
 初始化一个:NewDiskQueue 会定时执行:
 func (d *DiskQueue) ioLoop() {
    var dataRead []byte
    var err error
    var count int64
    var r chan []byte

    syncTicker := time.NewTicker(d.syncTimeout)

    for {
        count++
        // dont sync all the time :)
        if count == d.syncEvery {
            count = 0
            d.needSync = true
        }

        if d.needSync {
            err = d.sync()
            if err != nil {
                log.Printf("ERROR: diskqueue(%s) failed to sync - %s", d.name, err.Error())
            }
        }

        if (d.readFileNum < d.writeFileNum) || (d.readPos < d.writePos) {
            if d.nextReadPos == d.readPos {
                dataRead, err = d.readOne()
                if err != nil {
                    log.Printf("ERROR: reading from diskqueue(%s) at %d of %s - %s",
                        d.name, d.readPos, d.fileName(d.readFileNum), err.Error())
                    d.handleReadError()
                    continue
                }
            }
            r = d.readChan
        } else {
            r = nil
        }

        select {
        // the Go channel spec dictates that nil channel operations (read or write)
        // in a select are skipped, we set r to d.readChan only when there is data to read
        case r <- dataRead:
            d.moveForward()
        case <-d.emptyChan:
            d.emptyResponseChan <- d.deleteAllFiles()
        case dataWrite := <-d.writeChan:  //writeChan   刚刚写入数据的chan
            d.writeResponseChan <- d.writeOne(dataWrite)
        case <-syncTicker.C:
            d.needSync = true
        case <-d.exitChan:
            goto exit
        }
    }

exit:
    log.Printf("DISKQUEUE(%s): closing ... ioLoop", d.name)
    syncTicker.Stop()
    d.exitSyncChan <- 1
}


共有 人打赏支持
粉丝 7
博文 152
码字总数 57940
×
昏鸦
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: