nsq 源码分析之tcp协议部分

    最近恰好看到其余几个项目有socket 编程，而后，想了下，在golang 中还没用过socket tcp 编程，因而看了一些im 的协议解析过程，都不是太满意，不是太通用，恰好看到nsq，发现nsq 这部分真是简单粗暴。还用的是标准库的一些东西，很是通用。

    该文章后续仍在不断的更新修改中， 请移步到原文地址http://dmwan.cc

    nsq 的协议文档地址：https://nsq.io/clients/tcp_protocol_spec.html。

    首先，tcp 编程，最麻烦的地方是处理粘包，这个玩意怎么处理，通常是经过分隔符和两次读处理。httppaser 基本会封装这两种，好比tornado 本身封装的readbytes，readuntil，一个是定长读，一个是读到某个标识符为止，中间会为读写分配个缓冲区。这里道理很好理解，定长读，先读约定的header，好比先读4个字节，获得后面有n个字节后，一直循环向buffer 放，而readuntil 每次read 放到buffer，不断find 分隔符。这里的边界条件，无论是python 仍是c 写起来都是比较麻烦的事情。

    其实golang 里算是比较简单的，看个例子，下面这个是典型，第一行是command，'\n'分隔符结尾，第二行，由于这个command 有参数，用4个字节约定后面body 长度。

Publish a message to a topic:

PUB <topic_name>\n
[ 4-byte size in bytes ][ N-byte binary data ]

<topic_name> - a valid string (optionally having #ephemeral suffix)

    咱们作的时候，只须要有两个函数，一个能readuntil，一个能readbytes就ok，golang 有没？其实golang标准库是有的，分别是bufio.ReadSlice ，对应readuntil, 一个是io.ReadFull, 对应readbytes。

    nsq tcp server  流程算是比较清晰的。

    1，tcp accept 后，由一个 handle 处理，每一个conn 一个IOLoop, 这个IOLoop 会维持这个连接，进行读写。

 2， IoLoop 为每一个conn 新建一个client 对象，这个对象的Reader 实例化的时候用的是*bufio.Reader，而bufio 封装了一系列的io 方法，这里比较核心的是ReadSlice ，能读到后缀为止。读到的line 按空格解析成command 和 参数 ， 由后面反射成不一样命令，由Exec 执行。

    3， p.Exec 会根据不一样的命令，执行不一样方法。

    4，有的命令会附带body ，这里body 按照协议，是由4 byte 的size 加具体body 构成。这里解析，分两次读取。以pub 为例：

    readlen 其实就是ReadFull，client.lenSlice 就是一个[4]byte 的字节数组，因此，其实很明显，两次读取，一个读长度，按长度分配buffer，二次读取body。

func readLen(r io.Reader, tmp []byte) (int32, error) {
	_, err := io.ReadFull(r, tmp)
	if err != nil {
		return 0, err
	}
	return int32(binary.BigEndian.Uint32(tmp)), nil
}

    上面过程基本就能够 理解整个nsq tcp server 的数据流向了,这里的标准库函数能够很快移植到其余的tcp server 项目中去，只须要构建好本身的protocol，本身的命令反射就ok。

    下面分析下 ReadFull 的源码，相比其余语言，按定长读取，放到buffer，这里仍是比较有意思的。这样，读到指定长度err 为nil，读不到，数据为bad data。同理，readslice 方式相似。

// ReadAtLeast reads from r into buf until it has read at least min bytes.
// It returns the number of bytes copied and an error if fewer bytes were read.
// The error is EOF only if no bytes were read.
// If an EOF happens after reading fewer than min bytes,
// ReadAtLeast returns ErrUnexpectedEOF.
// If min is greater than the length of buf, ReadAtLeast returns ErrShortBuffer.
// On return, n >= min if and only if err == nil.
func ReadAtLeast(r Reader, buf []byte, min int) (n int, err error) {
	if len(buf) < min {
		return 0, ErrShortBuffer
	}
	for n < min && err == nil {
		var nn int
		nn, err = r.Read(buf[n:])// 这里不会按定长读取1024，而是按照传参，变长读，循环放倒buf，不会出现读多的问题 
		n += nn
	}
	if n >= min {
		err = nil
	} else if n > 0 && err == EOF {
		err = ErrUnexpectedEOF
	}
	return
}

// ReadFull reads exactly len(buf) bytes from r into buf.
// It returns the number of bytes copied and an error if fewer bytes were read.
// The error is EOF only if no bytes were read.
// If an EOF happens after reading some but not all the bytes,
// ReadFull returns ErrUnexpectedEOF.
// On return, n == len(buf) if and only if err == nil.
func ReadFull(r Reader, buf []byte) (n int, err error) {
	return ReadAtLeast(r, buf, len(buf))
}

     如何防止client 端恶意不传完整参数？设置超时属性就ok。以上。