文档章节

nsqd 源码,写入数据

昏鸦
 昏鸦
发布于 2014/03/27 16:52
字数 375
阅读 120
收藏 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
}


© 著作权归作者所有

共有 人打赏支持
昏鸦
粉丝 6
博文 155
码字总数 58873
作品 0
程序员
两个nsqd间传输数据时,如果有一方或者双方都在LAN中,数据怎样传输?

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

天台道人 ⋅ 2015/04/02 ⋅ 1

nsq nsqd源码1

1: 启动类: main.go 利用:flag.NewFlagSet 解析传递的参数 '注册系统的信号量 exitChan := make(chan int) signalChan := make(chan os.Signal, 1) go func() { <-signalChan exitChan <......

昏鸦 ⋅ 2014/03/25 ⋅ 0

游戏服务器骨架--gonet2

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

xtaci ⋅ 2015/10/15 ⋅ 0

nsq在docker部署后nsqadmin无法resolve所有nsqd的host的解决方法

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

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

去中心化分布式服务实时消息平台-NSQ

NSQ是一个基于Go语言的开源的分布式实时消息平台 NSQ可用于大规模系统的实时消息服务,它的设计目标是为在分布式环境下提供一个强大的去除中心化的分布式服务架构,可以每天处理数以亿计的实...

kiwisoft ⋅ 2016/10/27 ⋅ 0

NSQ 0.3.1 发布,开源高性能消息系统

NSQ 0.3.1 发布了,该版本包含一些小 bug 修复和新特性。改进了 nsqstat 和 nsqtofile 助手应用(支持 #ephemeral 主题)。此外 TLS 选项继续改进对 --tls-min-version 的支持以及 TLSFALLB...

oschina ⋅ 2015/01/27 ⋅ 5

NSQ:分布式的实时消息平台

NSQ是一个基于Go语言的分布式实时消息平台,它基于MIT开源协议发布,代码托管在GitHub,其当前最新版本是0.3.1版。 NSQ可用于大规模系统中的实时消息服务,并且每天能够处理数亿级别的消息,...

hejunyi1204 ⋅ 2015/08/14 ⋅ 0

Docker部属Nsq集群

  用一了段时间NSQ还是很稳定的。除了稳定,还有一个特别值的说的就是部署非常简单。总想写点什么推荐给大家使用nsq来做一些东西。但是就是因为他太简单易用,文档也比较简单易懂。一直不知...

lpxxn ⋅ 2017/10/25 ⋅ 0

nsq 使用笔记

nsqd 出现问题 https://github.com/nsqio/nsq/issues/691 https://github.com/nsqio/nsq/issues/506 nsqadmin 在访问的时候报错误 这个错误就是DNS访问出了问题。 解决方法: 使用 -broadca...

solate ⋅ 2016/11/01 ⋅ 0

Web 安全扫描平台--Gryffin

Gryffin 是雅虎开发的一个大规模 Web 安全扫描平台。它不是另外一个扫描器,其主要目的是为了解决两个特定的问题 —— 覆盖率和伸缩性。 该平台采用 Go 语言开发,依赖: Go PhantomJS, v2 S...

红薯 ⋅ 2015/09/27 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

CENTOS7防火墙命令记录

安装Firewall命令: yum install firewalld firewalld-config Firewall开启常见端口命令: firewall-cmd --zone=public --add-port=80/tcp --permanent firewall-cmd --zone=public --add-po......

cavion ⋅ 51分钟前 ⋅ 0

【C++】【STL】利用chromo来测量程序运行时间与日志时间打印精确到微秒

直接上代码吧,没啥好说的。头疼。 #include <iostream>#include <string>#include <ctime>#include <sstream>#include <iomanip>#include <thread>#include <chrono>using ......

muqiusangyang ⋅ 54分钟前 ⋅ 0

Mac环境下svn的使用

在Windows环境中,我们一般使用TortoiseSVN来搭建svn环境。在Mac环境下,由于Mac自带了svn的服务器端和客户端功能,所以我们可以在不装任何第三方软件的前提下使用svn功能,不过还需做一下简...

故久呵呵 ⋅ 今天 ⋅ 0

破解公司回应苹果“USB限制模式”:已攻破

本周四,苹果发表声明称 iOS 中加入了一项名为“USB 限制模式”的功能,可以防止 iPhone 在连接其他设备的时候被破解,并且强调这一功能并不是针对 FBI 等执法部门,为的是保护用户数据安全。...

六库科技 ⋅ 今天 ⋅ 0

MyBtais整合Spring Boot整合,TypeHandler对枚举类(enum)处理

概要 问题描述 我想用枚举类来表示用户当前状态,枚举类由 code 和 msg 组成,但我只想把 code 保存到数据库,查询处理,能知道用户当前状态,这应该怎么做呢?在 Spring 整合MyBatis 的时候...

Wenyi_Feng ⋅ 今天 ⋅ 0

synchronized与Lock的区别

# <center>王梦龙的读书笔记第一篇</center> ## <center>-synchronized与Lock的区别</centre> ###一、从使用场景来说 + synchronized 是能够注释代码块、类、方法但是它的加锁是和解锁使用一......

我不想加班 ⋅ 今天 ⋅ 0

VConsole的使用

手机端控制台打印输出,方便bug的排查。 首先需要引入vconsole.min.js 文件,然后在文件中创造实例。就能直接使用了。 var vConsole = new VConsole(); vConsole的文件地址...

大美琴 ⋅ 今天 ⋅ 0

Java NIO之字符集

1 字符集和编解码的概念 首先,解释一下什么是字符集。顾名思义,就是字符的集合。它的初衷是把现实世界的符号映射为计算机可以理解的字节。比如我创造一个字符集,叫做sex字符集,就包含两个...

士别三日 ⋅ 今天 ⋅ 0

Spring Bean基础

1、Bean之间引用 <!--如果Bean配置在同一个XML文件中,使用local引用--><ref bean="someBean"/><!--如果Bean配置在不同的XML文件中,使用ref引用--><ref local="someBean"/> 其实两种......

霍淇滨 ⋅ 今天 ⋅ 0

05、基于Consul+Upsync+Nginx实现动态负载均衡

1、Consul环境搭建 下载consul_0.7.5_linux_amd64.zip到/usr/local/src目录 cd /usr/local/srcwget https://releases.hashicorp.com/consul/0.7.5/consul_0.7.5_linux_amd64.zip 解压consu......

北岩 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部