nsq (三) 消息传输的可靠性和持久化[一]

上两篇帖子主要说了一下nsq的拓扑结构，如何进行故障处理和横向扩展,保证了客户端和服务端的长连接，连接保持了，就要传输数据了，nsq如何保证消息被订阅者消费，如何保证消息不丢失，就是今天要阐述的内容。

nsq topic、channel、和消费我客户端的结构如上图，一个topic下有多个channel每个channel可以被多个客户端订阅。
消息处理的大概流程:当一个消息被nsq接收后，传给相应的topic,topic把消息传递给所有的channel ,channel根据算法选择一个订阅客户端，把消息发送给客户端进行处理。
看上去这个流程是没有问题的，我们来思考几个问题

  • 网络传输的不确定性，比如超时；客户端处理消息时崩溃等，消息如何重传；
  • 如何标识消息被客户端成功处理完毕；
  • 消息的持久化，nsq服务端重新启动时消息不丢失；

服务端对发送中的消息处理逻辑

之前的帖子说过客户端和服务端进行连接后，会启动一个gorouting来发送信息给客户端

    go p.messagePump(client, messagePumpStartedChan)

然后会监听客户端发过来的命令client.Reader.ReadSlice(‘\n‘)
服务端会定时检查client端的连接状态，读取客户端发过来的各种命令，发送心跳等。每一个连接最终的目的就是监听channel的消息，发送给客户端进行消费。
当有消息发送给订阅客户端的时候，当然选择哪个client也是有无则的，这个以后讲，

func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
    // ...
    for {
        // ...
        case b := <-backendMsgChan:
            if sampleRate > 0 && rand.Int31n(100) > sampleRate {
                continue
            }

            msg, err := decodeMessage(b)
            if err != nil {
                p.ctx.nsqd.logf(LOG_ERROR, "failed to decode message - %s", err)
                continue
            }
            msg.Attempts++

            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
            client.SendingMessage()
            err = p.SendMessage(client, msg)
            if err != nil {
                goto exit
            }
            flushed = false
        case msg := <-memoryMsgChan:
            if sampleRate > 0 && rand.Int31n(100) > sampleRate {
                continue
            }
            msg.Attempts++

            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
            client.SendingMessage()
            err = p.SendMessage(client, msg)
            if err != nil {
                goto exit
            }
            flushed = false
        case <-client.ExitChan:
            goto exit
        }
    }

// ...
}

看一下这个方法调用subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)，在发送给客户端之前，把这个消息设置为在飞翔中，

// pushInFlightMessage atomically adds a message to the in-flight dictionary
func (c *Channel) pushInFlightMessage(msg *Message) error {
    c.inFlightMutex.Lock()
    _, ok := c.inFlightMessages[msg.ID]
    if ok {
        c.inFlightMutex.Unlock()
        return errors.New("ID already in flight")
    }
    c.inFlightMessages[msg.ID] = msg
    c.inFlightMutex.Unlock()
    return nil
}

然后发送给客户端进行处理。
在发送中的数据，存在的各种不确定性，nsq的处理方式是：对发送给客户端信息设置为在飞翔中，如果在如果处理成功就把这个消息从飞翔中的状态中去掉，如果在规定的时间内没有收到客户端的反馈，则认为这个消息超时，然后重新归队，两次进行处理。所以无论是哪种特殊情况，nsq统一认为消息为超时。

服务端处理超时消息

nsq对超时消息的处理，借鉴了redis的过期算法，但也不太一样redis的更复杂一些，因为redis是单线程的，还要处理占用cpu时间等等，nsq因为gorouting的存在要很简单很多。
简单来说，就是在nsq启动的时候启动协程去处理channel的过期数据

func (n *NSQD) Main() error {
    // ...
    // 启动协程去处理channel的过期数据
    n.waitGroup.Wrap(n.queueScanLoop)
    n.waitGroup.Wrap(n.lookupLoop)
    if n.getOpts().StatsdAddress != "" {
        n.waitGroup.Wrap(n.statsdLoop)
    }

    err := <-exitCh
    return err
}

当然不是每一个channel启动一个协程来处理过期数据，而是有一些规定，我们看一下一些默认值，然后再展开讲算法

    return &Options{
        // ...

        HTTPClientConnectTimeout: 2 * time.Second,
        HTTPClientRequestTimeout: 5 * time.Second,
        // 内存最大队列数
        MemQueueSize:    10000,
        MaxBytesPerFile: 100 * 1024 * 1024,
        SyncEvery:       2500,
        SyncTimeout:     2 * time.Second,

        // 扫描channel的时间间隔
        QueueScanInterval:        100 * time.Millisecond,
        // 刷新扫描的时间间隔
        QueueScanRefreshInterval: 5 * time.Second,
        QueueScanSelectionCount:  20,
        // 最大的扫描池数量
        QueueScanWorkerPoolMax:   4,
        // 标识百分比
        QueueScanDirtyPercent:    0.25,
        // 消息超时
        MsgTimeout:    60 * time.Second,
        MaxMsgTimeout: 15 * time.Minute,
        MaxMsgSize:    1024 * 1024,
        MaxBodySize:   5 * 1024 * 1024,
        MaxReqTimeout: 1 * time.Hour,
        ClientTimeout: 60 * time.Second,

        // ...
    }

这些参数都可以在启动nsq的时候根据自己需要来指定，我们主要说一下这几个：

  • QueueScanWorkerPoolMax就是最大协程数，默认是4，这个数是扫描所有channel的最大协程数，当然channel的数量小于这个参数的话，就调整协程的数量，以最小的为准，比如channel的数量为2个，而默认的是4个，那就调扫描的数量为2个
  • QueueScanSelectionCount 每次扫描最大的channel数量，默认是20，如果channel的数量小于这个值，则以channel的数量为准。
  • QueueScanDirtyPercent 标识脏数据 channel的百分比，默认为0.25，eg: channel数量为10,则一次最多扫描10个，查看每个channel是否有过期的数据，如果有，则标记为这个channel是有脏数据的，如果有脏数据的channel的数量 占这次扫描的10个channel的比例超过这个百分比,则直接再次进行扫描一次，而不用等到下一次时间点。
  • QueueScanInterval 扫描channel的时间间隔，默认的是每100毫秒扫描一次。
  • QueueScanRefreshInterval 刷新扫描的时间间隔 目前的处理方式是调整channel的协程数量。
    这也就是nsq处理过期数据的算法，总结一下就是，使用协程定时去扫描随机的channel里是否有过期数据。
func (n *NSQD) queueScanLoop() {
    workCh := make(chan *Channel, n.getOpts().QueueScanSelectionCount)
    responseCh := make(chan bool, n.getOpts().QueueScanSelectionCount)
    closeCh := make(chan int)

    workTicker := time.NewTicker(n.getOpts().QueueScanInterval)
    refreshTicker := time.NewTicker(n.getOpts().QueueScanRefreshInterval)

    channels := n.channels()
    n.resizePool(len(channels), workCh, responseCh, closeCh)

    for {
        select {
        case <-workTicker.C:
            if len(channels) == 0 {
                continue
            }
        case <-refreshTicker.C:
            channels = n.channels()
            n.resizePool(len(channels), workCh, responseCh, closeCh)
            continue
        case <-n.exitChan:
            goto exit
        }

        num := n.getOpts().QueueScanSelectionCount
        if num > len(channels) {
            num = len(channels)
        }

    loop:
        // 随机channel
        for _, i := range util.UniqRands(num, len(channels)) {
            workCh <- channels[i]
        }

        numDirty := 0
        for i := 0; i < num; i++ {
            if <-responseCh {
                numDirty++
            }
        }

        if float64(numDirty)/float64(num) > n.getOpts().QueueScanDirtyPercent {
            goto loop
        }
    }

exit:
    n.logf(LOG_INFO, "QUEUESCAN: closing")
    close(closeCh)
    workTicker.Stop()
    refreshTicker.Stop()
}

在扫描channel的时候，如果发现有过期数据后，会重新放回到队列，进行重发操作。

func (c *Channel) processInFlightQueue(t int64) bool {
    // ...
    for {
        c.inFlightMutex.Lock()
        msg, _ := c.inFlightPQ.PeekAndShift(t)
        c.inFlightMutex.Unlock()

        if msg == nil {
            goto exit
        }
        dirty = true

        _, err := c.popInFlightMessage(msg.clientID, msg.ID)
        if err != nil {
            goto exit
        }
        atomic.AddUint64(&c.timeoutCount, 1)
        c.RLock()
        client, ok := c.clients[msg.clientID]
        c.RUnlock()
        if ok {
            client.TimedOutMessage()
        }
        //重新放回队列进行消费处理。
        c.put(msg)
    }

exit:
    return dirty
}

客户端对消息的处理和响应

之前的帖子中的例子中有说过，客户端要消费消息，需要实现接口

type Handler interface {
    HandleMessage(message *Message) error
}

在服务端发送消息给客户端后，如果在处理业务逻辑时，如果发生错误则给服务器发送Requeue命令告诉服务器，重新发送消息进处理。如果处理成功，则发送Finish命令

func (r *Consumer) handlerLoop(handler Handler) {
    r.log(LogLevelDebug, "starting Handler")

    for {
        message, ok := <-r.incomingMessages
        if !ok {
            goto exit
        }

        if r.shouldFailMessage(message, handler) {
            message.Finish()
            continue
        }

        err := handler.HandleMessage(message)
        if err != nil {
            r.log(LogLevelError, "Handler returned error (%s) for msg %s", err, message.ID)
            if !message.IsAutoResponseDisabled() {
                message.Requeue(-1)
            }
            continue
        }

        if !message.IsAutoResponseDisabled() {
            message.Finish()
        }
    }

exit:
    r.log(LogLevelDebug, "stopping Handler")
    if atomic.AddInt32(&r.runningHandlers, -1) == 0 {
        r.exit()
    }
}

服务端收到命令后，对飞翔中的消息进行处理，如果成功则去掉，如果是Requeue则执行归队和重发操作，或者进行defer队列处理。

消息的持久化

默认的情况下，只有内存队列不足时MemQueueSize:10000时，才会把数据保存到文件内进行持久到硬盘。

    select {
    case c.memoryMsgChan <- m:
    default:
        b := bufferPoolGet()
        err := writeMessageToBackend(b, m, c.backend)
        bufferPoolPut(b)
        c.ctx.nsqd.SetHealth(err)
        if err != nil {
            c.ctx.nsqd.logf(LOG_ERROR, "CHANNEL(%s): failed to write message to backend - %s",
                c.name, err)
            return err
        }
    }
    return nil

如果将 --mem-queue-size 设置为 0，所有的消息将会存储到磁盘。我们不用担心消息会丢失，nsq 内部机制保证在程序关闭时将队列中的数据持久化到硬盘，重启后就会恢复。
nsq自己开发了一个库go-diskqueue来持久会消息到内存。这个库的代码量不多，理解起来也不难,代码逻辑我想下一篇再讲。
看一下保存在硬盘后的样子：

原文地址：https://www.cnblogs.com/li-peng/p/11770451.html

时间： 2024-07-30 15:09:23

nsq (三) 消息传输的可靠性和持久化[一]的相关文章

RabbitMQ(三):消息持久化策略

原文:RabbitMQ(三):消息持久化策略 一.前言 在正常的服务器运行过程中,时常会面临服务器宕机重启的情况,那么我们的消息此时会如何呢?很不幸的事情就是,我们的消息可能会消失,这肯定不是我们希望见到的结果.所以我们希望AMQP服务器崩溃了也可以将消息恢复,这称之为消息持久化.RabbitMQ自然存在这种策略可以帮助我们完成这件事情. 二.持久化的消息 当RabbitMQ服务器重启后,原先的队列和交换器会随同里面的消息一同消失.原因在于每个队列和交换器都有durable属性,该属性默认是fa

Rsyslog的三种传输协议简要介绍

rsyslog的三种传输协议 rsyslog 可以理解为多线程增强版的syslog. rsyslog提供了三种远程传输协议,分别是: 1. UDP 传输协议 基于传统UDP协议进行远程日志传输,也是传统syslog使用的传输协议: 可靠性比较低,但性能损耗最少, 在网络情况比较差, 或者接收服务器压力比较高情况下,可能存在丢日志情况. 在对日志完整性要求不是很高,在可靠的局域网环境下可以使用. 2.TCP 传输协议 基于传统TCP协议明文传输,需要回传进行确认,可靠性比较高: 但在接收服务器宕机

实时消息传输协议 RTMP(Real Time Messaging Protocol)

原文: http://blog.csdn.net/defonds/article/details/17403225 译序:本文是维基百科关于 RTMP 的解释, 关于 RTMP 官方规范参见 RTMP 规范,关于 RTMP 官方规范的中文版,参见<Adobe 官方公布的 RTMP 规范>.以下是维基百科原文: 实时消息传输协议(RTMP)最初是由 Macromedia 为互联网上 Flash player 和服务器之间传输音频.视频以及数据流而开发的一个私有协议.Adobe 收购 Macrom

WCF 客户端与服务端消息传输

WCF很多需要认证信息,保证服务的安全,可以使用消息来实现 WCF 实现消息的方式: WCF中有两个接口: IClientMessageInspector [定义一个消息检查器对象,该对象可以添加到 System.ServiceModel.Dispatcher.ClientRuntime.MessageInspectors集合来查看或修改消息] IDispatchMessageInspector  [定义一些方法,通过这些方法,可以在服务应用程序中对入站和出站应用程序消息进行自定义检查或修改.]

ZeroMQ接口函数之 ：zmq - 0MQ 轻量级消息传输内核

zmq(7) 0MQ Manual - 0MQ/3.2.5 Name zmq – ØMQ 轻量级消息传输内核 Synopsis #include <znq.h> cc [flags] files –lzmq [libraries] Description ØMQ轻量级消息传输内核是一个从标准socket接口的扩展而来的链接库,这些接口通常是由一些专门的传送中间设备来提供.ØMQ提供了一个步消息传送.多模式消息传送.消息过滤(订阅).对多种传输协议无缝接入的集合. 本文档呈现了ØMQ的概念,描述

医疗行业多层级复杂网络环境下的消息传输（远程会诊）架构与实现

近期接手一个针对医疗系统远程会诊平台的技术改造工作,这项工作中的一些技术问题颇具代表性,我会在此记录这一工作的过程和技术细节,如果条件允许,会在 GitHub 上开源部分业务无关的纯技术实现,敬请关注.(https://github.com/iccb1013). 远程会诊平台的应用场景指的是乡镇或县卫生所,在接诊过程中,对疑难问题上报上级医疗机构,由上级医疗机构进行网络诊断并回复诊疗意见,但是这一过程,并不是简单的点对点的关系.主要特点:1)  包含多级机构:乡镇.县.区.市.省.由任意一级向上

企业级工作流解决方案（七）--微服务Tcp消息传输模型之消息编解码

Tcp消息传输主要参照surging来做的,做了部分裁剪和改动,详细参见:https://github.com/dotnetcore/surging Json-rpc没有定义消息如何传输,因此,Json-Rpc RpcRequest对象和RpcResponse对象需要一个传输载体,这里的传输对象主是TransportMessage,如下代码,这里的Content请求时为RcpRequest对象,答复时为RpcResponse对象,答复时Header一般情况下为空. /// <summary>

RabbitMQ原理与相关操作(三)消息持久化

现在聊一下RabbitMQ消息持久化: 问题及方案描述 1.当有多个消费者同时收取消息,且每个消费者在接收消息的同时,还要处理其它的事情,且会消耗很长的时间.在此过程中可能会出现一些意外,比如消息接收到一半的时候,一个消费者死掉了. 这种情况要使用消息接收确认机制,可以执行上次宕机的消费者没有完成的事情. 2.在默认情况下,我们程序创建的消息队列以及存放在队列里面的消息,都是非持久化的.当RabbitMQ死掉了或者重启了,上次创建的队列.消息都不会保存. 这种情况可以使用RabbitMQ提供的消

漫游Kafka设计篇之消息传输的事务定义

之前讨论了consumer和producer是怎么工作的,现在来讨论一下数据传输方面.数据传输的事务定义通常有以下三种级别: 最多一次: 消息不会被重复发送,最多被传输一次,但也有可能一次不传输. 最少一次: 消息不会被漏发送,最少被传输一次,但也有可能被重复传输. 精确的一次(Exactly once):  不会漏传输也不会重复传输,每个消息都传输被一次而且仅仅被传输一次,这是大家所期望的. 大多数消息系统声称可以做到“精确的一次”,但是仔细阅读它们的的文档可以看到里面存在误导,比如没有说明当