golang-redis系列——返回值助手函数(二)

从上一节的内容可知，Do() 和 Receive() 等方法的返回值，除了 error 外，是一个 interface{} 类型的返回值，因此当我们的复杂操作返回的不是基本数据类型时，就需要我们自己解析返回值，例如，当我们利用 HMGET 方法获取一批返回值时，就需要对返回结果进行解析，具体如下：

创新互联-专业网站定制、快速模板网站建设、高性价比东平网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式东平网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖东平地区。费用合理售后完善，10年实体公司更值得信赖。

由于返回值是多条数据，因此需要先将 reply 转成 []interface 类型，然后在遍历结果时在分别转成 []uint8 (byte数组)， 最后再转成 string 类型。

随着我们操作复杂度，数据解析的工作量也会非常大，(lua 脚本的使用，会使结果的解析更为复杂，因为可能存在多种类型的结果一起返回的情况，lua 脚本相关的内容会在下一节介绍)。

redigo 包中的返回值助手函数的存在，就是为了帮助我们完成这些枯燥繁琐的数据解析过程。

返回值助手函数相关源码路径为 github.com/gomodule/redigo/redis/reply.go 提供的主要方法如下：

上述返回值助手函数的具体使用，应该依据具体的命令进行选择。如果大家还记得上一节介绍的 Redis 基本数据类型，可能会有些疑问，对于 redis 来说，其数据据存储本质都是 []bytes， 为什么可以解析出 Int、int64、float等类型的数据呢？

我们以 Float64() 为例进行说明，具体源码如下：

其实，返回值助手函数是将 []byte 类型的原始数据，利用 strconv.ParseFloat(string(reply), 64) 转换成了 float64类型，因此在我们使用过程中返回值助手函数的选择，应该基于业务和实际存储的数据格式为依据。我们以第一小节的示例为例，看返回值助手函数如何降低我们的工作量，具体如下：

除了使用返回值助手函数对上述固定结构的结果进行解析外，redigo 包还提供了一个 Scan()函数用于解析自定义的复杂数据结构，我们依然以上一个示例进行说明,具体示例如下：

如果返回结果为结构化切片，也可以使用 canSlice() 方法，从而简化 loop 处理的部分，具体示例如下：

通过上述的示例，我们介绍了 scan 函数的基本用法，但是细心的同学可能会发现吗，为什么数据写入时，value 的类型为 []int64 但是读取时只能按照 string 类型读取呢。这是因为 Redis 底层存储的数据本质都是 string 类型，。 无论是 HMSET 还是 MSET 最终都只能按照 string 类型读取，因为其本质都是 hash 结构，不同之处仅在于 HMSET 是嵌套的 hash类型。 因此，[]int64 数据在写入阶段，就已经被自动处理为 []byte，写入 redis 之后，len 和 类型 属性会丢失。

如果强行按照 []int64解析将出错：

如果 value 必须以结构化的数据存储，那么可以提前对要写入的数据进行编码，例如 json、protobuf 等，取出后再进行解码获得原始数据。

详解golang中bufio包的实现原理

最近用golang写了一个处理文件的脚本，由于其中涉及到了文件读写，开始使用golang中的 io 包，后来发现golang 中提供了一个bufio的包，使用这个包可以大幅提高文件读写的效率，于是在网上搜索同样的文件读写为什么bufio 要比io的读写更快速呢？根据网上的资料和阅读源码，以下来详细解释下bufio的高效如何实现的。

bufio 包介绍 

bufio包实现了有缓冲的I/O。它包装一个io.Reader或io.Writer接口对象，创建另一个也实现了该接口，且同时还提供了缓冲和一些文本I/O的帮助函数的对象。

以上为官方包的介绍，在其中我们能了解到的信息如下：

bufio 是通过缓冲来提高效率

简单的说就是，把文件读取进缓冲（内存）之后再读取的时候就可以避免文件系统的io 从而提高速度。同理，在进行写操作时，先把文件写入缓冲（内存），然后由缓冲写入文件系统。看完以上解释有人可能会表示困惑了，直接把 内容-文件 和 内容-缓冲-文件相比， 缓冲区好像没有起到作用嘛。其实缓冲区的设计是为了存储多次的写入，最后一口气把缓冲区内容写入文件。下面会详细解释

bufio 封装了io.Reader或io.Writer接口对象，并创建另一个也实现了该接口的对象

io.Reader或io.Writer 接口实现read() 和 write() 方法，对于实现这个接口的对象都是可以使用这两个方法的

bufio 包实现原理

bufio 源码分析

Reader对象

bufio.Reader 是bufio中对io.Reader 的封装

// Reader implements buffering for an io.Reader object.

type Reader struct {

buf     []byte

rd      io.Reader // reader provided by the client

r, w     int    // buf read and write positions

err     error

lastByte   int

lastRuneSize int

}

bufio.Read(p []byte) 相当于读取大小len(p)的内容，思路如下：

当缓存区有内容的时，将缓存区内容全部填入p并清空缓存区

当缓存区没有内容的时候且len(p)len(buf),即要读取的内容比缓存区还要大，直接去文件读取即可

当缓存区没有内容的时候且len(p)len(buf),即要读取的内容比缓存区小，缓存区从文件读取内容充满缓存区，并将p填满（此时缓存区有剩余内容）

以后再次读取时缓存区有内容，将缓存区内容全部填入p并清空缓存区（此时和情况1一样）

以下是源码

// Read reads data into p.

// It returns the number of bytes read into p.

// The bytes are taken from at most one Read on the underlying Reader,

// hence n may be less than len(p).

// At EOF, the count will be zero and err will be io.EOF.

func (b *Reader) Read(p []byte) (n int, err error) {

n = len(p)

if n == 0 {

return 0, b.readErr()

}

if b.r == b.w {

if b.err != nil {

return 0, b.readErr()

}

if len(p) = len(b.buf) {

// Large read, empty buffer.

// Read directly into p to avoid copy.

n, b.err = b.rd.Read(p)

if n 0 {

panic(errNegativeRead)

}

if n 0 {

b.lastByte = int(p[n-1])

b.lastRuneSize = -1

}

return n, b.readErr()

}

// One read.

// Do not use b.fill, which will loop.

b.r = 0

b.w = 0

n, b.err = b.rd.Read(b.buf)

if n 0 {

panic(errNegativeRead)

}

if n == 0 {

return 0, b.readErr()

}

b.w += n

}

// copy as much as we can

n = copy(p, b.buf[b.r:b.w])

b.r += n

b.lastByte = int(b.buf[b.r-1])

b.lastRuneSize = -1

return n, nil

}

说明：

reader内部通过维护一个r, w 即读入和写入的位置索引来判断是否缓存区内容被全部读出

Writer对象

bufio.Writer 是bufio中对io.Writer 的封装

// Writer implements buffering for an io.Writer object.

type Writer struct {

err error

buf []byte

n  int

wr io.Writer

}

bufio.Write(p []byte) 的思路如下

判断buf中可用容量是否可以放下 p

如果能放下，直接把p拼接到buf后面，即把内容放到缓冲区

如果缓冲区的可用容量不足以放下，且此时缓冲区是空的，直接把p写入文件即可

如果缓冲区的可用容量不足以放下，且此时缓冲区有内容，则用p把缓冲区填满，把缓冲区所有内容写入文件，并清空缓冲区

判断p的剩余内容大小能否放到缓冲区，如果能放下（此时和步骤1情况一样）则把内容放到缓冲区

如果p的剩余内容依旧大于缓冲区，（注意此时缓冲区是空的，情况和步骤2一样）则把p的剩余内容直接写入文件

// Write writes the contents of p into the buffer.

// It returns the number of bytes written.

// If nn len(p), it also returns an error explaining

// why the write is short.

func (b *Writer) Write(p []byte) (nn int, err error) {

for len(p) b.Available() b.err == nil {

var n int

if b.Buffered() == 0 {

// Large write, empty buffer.

// Write directly from p to avoid copy.

n, b.err = b.wr.Write(p)

} else {

n = copy(b.buf[b.n:], p)

b.n += n

b.flush()

}

nn += n

p = p[n:]

}

if b.err != nil {

return nn, b.err

}

n := copy(b.buf[b.n:], p)

b.n += n

nn += n

return nn, nil

}

说明：

b.wr 存储的是一个io.writer对象，实现了Write()的接口，所以可以使用b.wr.Write(p) 将p的内容写入文件

b.flush() 会将缓存区内容写入文件，当所有写入完成后，因为缓存区会存储内容，所以需要手动flush()到文件

b.Available() 为buf可用容量，等于len(buf) - n

下图解释的是其中一种情况，即缓存区有内容，剩余p大于缓存区

golang p2p网

继续进入下一个初始化

n.netService, err = nebnet.NewNebService(n)

if err != nil {

logging.CLog().WithFields(logrus.Fields{

"err": err,

}).Fatal("Failed to setup net service.")

}

netservice有两个成员

type NebServicestruct {

node      *Node

dispatcher *Dispatcher

}

跳出stup()函数

先进入start()函数看一看

if err := n.netService.Start(); err != nil {

logging.CLog().WithFields(logrus.Fields{

"err": err,

}).Fatal("Failed to start net service.")

}

进入netservice.start()

func (ns *NebService) Start() error {

logging.CLog().Info("Starting NebService...")

// start dispatcher.

ns.dispatcher.Start()

// start node.

if err := ns.node.Start(); err != nil {

ns.dispatcher.Stop()

logging.CLog().WithFields(logrus.Fields{

"err": err,

}).Error("Failed to start NebService.")

return err

}

logging.CLog().Info("Started NebService.")

return nil

}

可以看到第一个start()的函数是dispatcher.start()

进入dispatch.start()

func (dp *Dispatcher) Start() {

logging.CLog().Info("Starting NebService Dispatcher...")

go dp.loop()

}

然后就出现一个新的线程、goruntime

go dp.loop()

进入该线程，看它干了些什么

timerChan := time.NewTicker(time.Second).C

for {

select {

case -timerChan:

metricsDispatcherCached.Update(int64(len(dp.receivedMessageCh)))

case -dp.quitCh:

logging.CLog().Info("Stoped NebService Dispatcher.")

return

case msg := -dp.receivedMessageCh:

msgType := msg.MessageType()

v, _ := dp.subscribersMap.Load(msgType)

if v == nil {

continue

  }

m, _ := v.(*sync.Map)

m.Range(func(key, valueinterface{}) bool {

select {

case key.(*Subscriber).msgChan - msg:

default:

logging.VLog().WithFields(logrus.Fields{

"msgType": msgType,

}).Warn("timeout to dispatch message.")

}

return true

  })

}

}

一个有点长的循环

metricsDispatcherCached.Update(int64(len(dp.receivedMessageCh)))一秒钟刷新一次缓冲区

case msg := -dp.receivedMessageCh:

msgType := msg.MessageType()如果能取出dp.receivedMessageCh

msgType := msg.MessageType()首先判断取出的信息类型

v, _ := dp.subscribersMap.Load(msgType)

if v == nil {

continue

}

根据类型取出相应的map

如果取不出，那么使用continue结束这个case

m, _ := v.(*sync.Map)

断言

m.Range(func(key, valueinterface{}) bool {

select {

case key.(*Subscriber).msgChan - msg:

default:

logging.VLog().WithFields(logrus.Fields{

"msgType": msgType,

}).Warn("timeout to dispa+tch message.")

}

return true

})

将msg推入其他管道里面去。其他goruntime会循环等待该

C语言goto语句问题:goto loop与goto repeat有什么区别吗？

没有区别goto

是跳转语句,后面是标识符,自己命名的.没什么特别含义,他这样写是为了区分,和让程序更加可读(这里可读是指标识符相对于无意义的字符,并不是说goto语句会是程序可读性更好,实际刚好相反).goto

loop从语义上理解应该是跳转到某一个循环里面让人知道跳到那个地方去(即前面所谓的可读好)

goto格式是

asd:xxxxxx

xxxxxxxx

xxxxxxx

goto

asd;

就是跳到asd重新在运行一次的意思,如果上面xxx中没有结束或者跳出的语句上面就会变成死循环.和

while(1)一个意思

一般程序中很少或根本不使用goto语句的,用了会很大的破坏可读性,和带来维护上的困难,所以不推荐使用

C语言loop: 这个什么意思？goto loop;

这个例子来说吧：loop: A语句;...(其他语句)goto loop;则goto loop表示程序转去执行loop标记的那个语句！

协程与异步IO

协程，又称微线程，纤程。英文名 Coroutine 。Python对协程的支持是通过 generator 实现的。在generator中，我们不但可以通过for循环来迭代，还可以不断调用 next()函数 获取由 yield 语句返回的下一个值。但是Python的yield不但可以返回一个值，它还可以接收调用者发出的参数。yield其实是终端当前的函数，返回给调用方。python3中使用yield来实现range，节省内存，提高性能，懒加载的模式。

asyncio是Python 3.4 版本引入的 标准库 ，直接内置了对异步IO的支持。

从Python 3.5 开始引入了新的语法 async 和 await ，用来简化yield的语法：

import asyncio

import threading

async def compute(x, y):

print("Compute %s + %s ..." % (x, y))

print(threading.current_thread().name)

await asyncio.sleep(x + y)

return x + y

async def print_sum(x, y):

result = await compute(x, y)

print("%s + %s = %s" % (x, y, result))

print(threading.current_thread().name)

if __name__ == "__main__":

loop = asyncio.get_event_loop()

tasks = [print_sum(1, 2), print_sum(3, 4)]

loop.run_until_complete(asyncio.wait(tasks))

loop.close()

线程是内核进行抢占式的调度的，这样就确保了每个线程都有执行的机会。而 coroutine 运行在同一个线程中，由语言的运行时中的 EventLoop（事件循环） 来进行调度。和大多数语言一样，在 Python 中，协程的调度是非抢占式的，也就是说一个协程必须主动让出执行机会，其他协程才有机会运行。

让出执行的关键字就是 await。也就是说一个协程如果阻塞了，持续不让出 CPU，那么整个线程就卡住了，没有任何并发。

PS: 作为服务端，event loop最核心的就是IO多路复用技术，所有来自客户端的请求都由IO多路复用函数来处理;作为客户端，event loop的核心在于利用Future对象延迟执行，并使用send函数激发协程,挂起,等待服务端处理完成返回后再调用CallBack函数继续下面的流程

Go语言的协程是 语言本身特性 ，erlang和golang都是采用了CSP(Communicating Sequential Processes)模式(Python中的协程是eventloop模型)，但是erlang是基于进程的消息通信，go是基于goroutine和channel的通信。

Python和Go都引入了消息调度系统模型，来避免锁的影响和进程/线程开销大的问题。

协程从本质上来说是一种用户态的线程，不需要系统来执行抢占式调度，而是在语言层面实现线程的调度 。因为协程 不再使用共享内存/数据 ，而是使用 通信 来共享内存/锁，因为在一个超级大系统里具有无数的锁，共享变量等等会使得整个系统变得无比的臃肿，而通过消息机制来交流，可以使得每个并发的单元都成为一个独立的个体，拥有自己的变量，单元之间变量并不共享，对于单元的输入输出只有消息。开发者只需要关心在一个并发单元的输入与输出的影响，而不需要再考虑类似于修改共享内存/数据对其它程序的影响。

网站名称：go语言的loop,go语言的特性
路径分享：http://kswjz.com/article/hodjhd.html