文档章节

nsqd 源码,写入数据

昏鸦
 昏鸦
发布于 2014/03/27 16:52
字数 375
阅读 122
收藏 1

切记: 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
}


© 著作权归作者所有

共有 人打赏支持
上一篇: nsq 流程图
昏鸦
粉丝 7
博文 184
码字总数 59972
作品 0
程序员
私信 提问
两个nsqd间传输数据时,如果有一方或者双方都在LAN中,数据怎样传输?

@昏鸦 你好,想跟你请教个问题: 按照我的学习了解:客户端通过查询nsqlookup得知自己想要的topic的新数据在哪些nsqd上,得到一个nsqd列表,然后客户端去连接这些nsqd,拿取它要的数据。 但是...

天台道人
2015/04/02
487
1
nsq源码分析之nsqlookup实现

nsqlookup服务为nsqd的服务发现,分析和借鉴下服务发现的源码: 再来看看nsqlookup是怎么去存储管理这些Producer: 分别看看怎么新增Registration和Producer: 再看看Find相应的Registration和P...

smart_w
2018/07/31
0
0
游戏服务器骨架--gonet2

欢迎使用 gonet/2是新一代游戏服务器骨架,基于go语言开发,采用了先进的http/2作为服务器端主要通信协议,以microservice作为主要思想进行架构,采用docker作为服务发布手段。相比第一代gon...

xtaci
2015/10/15
3.2K
0
深入NSQ 之旅

介绍 NSQ是一个实时的分布式消息平台。它的设计目标是为在多台计算机上运行的松散服务提供一个现代化的基础设施骨架。 这篇文章介绍了 基于go语言的NSQ的内部架构,它能够为高吞吐量的网络服...

老盖
2014/01/03
19.1K
9
nsq在docker部署后nsqadmin无法resolve所有nsqd的host的解决方法

nsq在docker部署后nsqadmin无法resolve所有nsqd的host的解决方法 疯子的自留地2017-12-173 阅读 docker方法部署 问题 最开始部署 nsq 的时候是直接在云主机上直接部署的,但是作为一个牛逼的程...

疯子的自留地
2017/12/17
0
0

没有更多内容

加载失败,请刷新页面

加载更多

MySQL学习笔记之二

数据库的操作总结就是:增删改查(CURD),今天记录一下基础的检索查询工作。 检索MySQL 1.查询表中所有的记录 mysql> select * from apps;+----+------------+-----------------------+------...

凌宇之蓝
30分钟前
0
0
PaddlePaddle-GitHub的正确打开姿势

GitHub是一个面向开源及私有软件项目的托管平台、也是项目版本管理工具,会使用它是程序员入门的必备技能。PaddlePaddle也不例外,所有的源码及项目进展都在GitHub上开源公布。但对于刚入门写...

深度学习之路
30分钟前
1
0
最强NLP模型BERT可视化学习

摘要: 最强NLP模型谷歌BERT狂破11项纪录,全面超越人类,本文通过可视化带你直观了解它。 2018年是自然语言处理(Natural Language Processing, NLP)领域的转折点,一系列深度学习模型在智...

阿里云官方博客
38分钟前
1
0
导出功能

public void downloadD(HttpServletRequest request, HttpServletResponse res,String contractName, String contractPath) throws IOException {// FileAttach fileAttach = fileA......

卖星星的小矮人
41分钟前
1
0
gradle 打包可执行jar包

group 'android.com'version '1.0-SNAPSHOT'apply plugin: 'java'sourceCompatibility = 1.8repositories { mavenCentral()}jar { manifest { attributes ('Main-......

zdglf
53分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部