go微服务系列之三

在前两篇系列博文中，我已经实现了user-srv、web-srv、api-srv，在新的一篇博文中，我要讲解的是如何在项目中如何使用redis存储session。如果想直接查阅源码或者通过demo学习的，可以访问ricoder_demo。

如何编写一个微服务？这里用的是go的微服务框架go micro，具体的情况可以查阅：http://btfak.com/%E5%BE%AE%E6%9C%8D%E5%8A%A1/2016/03/28/go-micro/

一、构建user-status.srv

1.1 构建UserStatus.proto

定义User的状态操作函数，部分源码如下：

syntax = "proto3";

package pb;

service UserStatus {
//通过uid获取session
rpc GetSessionByUID(GetSessionByUIDReq) returns (GetSessionByUIDRep) {}
//通过token获取session
rpc GetSessionByToken(GetSessionByTokenReq) returns (GetSessionByTokenRep) {}
//获取用户的长连接地址
rpc GetConnectorAddr(GetConnectorAddrReq) returns (GetConnectorAddrRep) {}
//更新用户长连接地址（用户建立长连接时调用)
rpc UpdateConnectorAddr(UpdateConnectorAddrReq) returns (UpdateConnectorAddrRep) {}
//构建session用户登录时调用，此接口会清除旧session
rpc NewSession(NewSessionReq) returns (NewSessionRep) {}
//移除session登出时会调用
rpc RemoveSession(RemoveSessionReq) returns (RemoveSessionRep) {}
//token续期
rpc RefreshSession(RefreshSessionReq) returns (RefreshSessionRep) {}
//更新用户长连接地址（用户建立长连接时调用)
rpc UserConnected(UserConnectedReq) returns (UserConnectedRep) {}
//删除用户的长连接地址（用户长连接断开时调用）
rpc UserDisonnected(UserDisonnectedReq) returns (UserDisonnectedRep) {}
//通过uid来移除session
rpc RemoveSessionByUID(RemoveSessionByUIDReq) returns (RemoveSessionByUIDRep) {}
//通过token找uid
rpc GetUserIDByToken(GetUserIDByTokenReq) returns (GetUserIDByTokenRep) {}
}
/*
还有一些定义，完整示例可以查看源码～
*/

1.2 运行脚本build_proto.sh自动构建userStatus.pb.go

$ bash ./build_proto.sh

这个build_proto.sh是我自己构建的一个脚本文件，运行之后会在/src/share/pb/文件夹下面生成一个userStatus.pb.go文件

1.3 构建handler，实现userStatus中的函数

我在src文件夹下面添加一个user-status-srv文件夹，并在里边添加一个handler文件夹和utils文件夹，一个存放handler文件，一个存放工具类函数，然后实现handler函数，源码如下：

package handler

import (
    //多个导入包，具体请查看源码
)
type UserStatusHandler struct {
    pool               *redis.Pool
    logger             *zap.Logger
    namespace          string
    sessionExpire int
    tokenExpire int
}

func NewUserStatusHandler(pool *redis.Pool) *UserStatusHandler {
    return &UserStatusHandler{
        pool: pool,
        sessionExpire: 15 * 86400,
        tokenExpire:   15 * 86400,
    }
}

//GetUserIDByToken GetUIDByToken
func (s *UserStatusHandler) GetUserIDByToken(ctx context.Context, req *pb.GetUserIDByTokenReq, rsp *pb.GetUserIDByTokenRep) error {
    return nil  
}
/*
还有其他函数的实现，完整示例可以查看源码～
*/

这里实现的函数全部先采用空实现，在后面会慢慢添加

1.4 实现main函数，启动service

源码如下：

package main

import (
    //多个导入包，具体查看完整源码
)

func main() {

    // 创建Service，并定义一些参数
    service := micro.NewService(
        micro.Name(config.Namespace+"userStatus"),
        micro.Version("latest"),
    )
    // 定义Service动作操作
    service.Init(
        micro.Action(func(c *cli.Context) {
            log.Println("micro.Action test ...")
            // 注册redis
            redisPool := share.NewRedisPool(3, 3, 1,300*time.Second,":6379","redis")
            // 先注册db
            db.Init(config.MysqlDSN)
            pb.RegisterUserStatusHandler(service.Server(), handler.NewUserStatusHandler(redisPool), server.InternalHandler(true))
        }),
        micro.AfterStop(func() error {
            log.Println("micro.AfterStop test ...")
            return nil
        }),
        micro.AfterStart(func() error {
            log.Println("micro.AfterStart test ...")
            return nil
        }),
    )

    log.Println("启动user-status-srv服务 ...")

    //启动service
    if err := service.Run(); err != nil {
        log.Panic("user-status-srv服务启动失败 ...")
    }
}

由源码可以看出，我在启动service之前先注册了redis、db以及绑定handler，再通过Run启动service。

1.5 查看consul

在浏览器打开 http://127.0.0.1:8500/ ，如果可以在页面中看到对应的srv，则说明service启动成功。如：

Screenshot from 2017-10-17 18-04-06.png

二、使用redis

在这一章节中，我将采用redis实现数据的存取。

2.1 新建一个redis.Pool

在main.go函数中，我使用 *share.NewRedisPool(3, 3, 1,300time.Second,":6379","redis") 得到了一个redisPool，NewRedisPool源码如下：

func NewRedisPool(maxIdle, maxActive , DBNum int, timeout time.Duration, addr , password string) *redis.Pool {

    return &redis.Pool{
        MaxActive:   maxActive,
        MaxIdle:     maxIdle,
        IdleTimeout: timeout,
        Wait:        true,
        Dial: func() (redis.Conn, error) {
            // return redis.DialURL(rawurl)
            // return redis.Dial("tcp", addr, redis.DialPassword(password), redis.DialDatabase(dbNum))
            return redis.Dial("tcp", addr, redis.DialPassword(password), redis.DialDatabase(DBNum))
        },
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            _, err := c.Do("PING")
            return err
        },
    }
}

在这里我使用的是第三方开源框架，有兴趣的可以查看 https://github.com/garyburd/redigo 了解情况。

2.2 使用redis存取数据

在这里我以NewSession为例，源码如下：

func (s *UserStatusHandler) NewSession(ctx context.Context, req *pb.NewSessionReq, rsp *pb.NewSessionRep) error {
    var oldSession *pb.Session
    defer func() {
        utils.SessionFree(oldSession)
    }()
    fieldMap := make(map[string]interface{}, 0)
    fieldMap["Uid"] = req.Id
    fieldMap["Address"] = req.Address
    fieldMap["Phone"] = req.Phone
    fieldMap["Name"] = req.Name
    //生成Token
    token, err := utils.NewToken(req.Id)
    if err != nil {
        log.Println("生成token失败", zap.Error(err), zap.Int32("uid", req.Id))
        return err
    }

    //删除所有旧token
    if err = utils.RemoveUserSessions(req.Id, s.pool); err != nil {
        log.Println("删除所有旧token失败", zap.Error(err), zap.Int32("uid", req.Id))
        return err
    }
    conn := s.pool.Get()
    //会话数据写入redis，格式：t:id => map的哈希值
    if _, err := conn.Do("HMSET", redis.Args{}.Add(utils.KeyOfSession(req.Id)).AddFlat(fieldMap)...); err != nil {
        conn.Close()
        log.Println("会话数据写入redis失败", zap.Error(err), zap.String("key", utils.KeyOfSession(req.Id)), zap.Any("参数", fieldMap))
        return err
    }
    //设置t:id的过期时间
    if _, err := conn.Do("EXPIRE", utils.KeyOfSession(req.Id), s.sessionExpire); err != nil {
        conn.Close()
        s.logger.Error("设置session过期时间失败", zap.Error(err), zap.String("key", utils.KeyOfSession(req.Id)))
        return err
    }

    //用户token写入set里边，格式：t:uid:set:id => token
    keyOfSet := utils.KeyOfSet(req.Id)
    if _, err = conn.Do("SADD", keyOfSet, token); err != nil {
        conn.Close()
        log.Println("token写入用户集合失败", zap.Error(err), zap.String("key", keyOfSet), zap.String("参数", token))
        return err
    }
    //设置t:uid:set:id的过期时间
    if _, err = conn.Do("EXPIRE", keyOfSet, s.sessionExpire); err != nil {
        conn.Close()
        log.Println("设置用户token集合过期时间失败", zap.Error(err), zap.String("key", keyOfSet))
        return err
    }

    //将token和id对应，格式：token => id
    if _, err = conn.Do("SETEX", utils.KeyOfToken(token), s.tokenExpire, req.Id); err != nil {
        conn.Close()
        log.Println("token写入redis失败", zap.Error(err), zap.String("key", utils.KeyOfToken(token)), zap.Int32("参数", req.Id))
        return err
    }

    rsp.Token = token
    return nil
}

如代码所示，操作redis的步骤是 conn := s.pool.Get() 先开启一个连接，再通过conn.Do("EXPIRE", keyOfSet, s.sessionExpire) 的一种方式操作redis中的数据，具体的可以查看redis的api，这里有个函数 utils.SessionFree(oldSession) ，这是我在utils包下自定义的一个函数，这个知识点再接下来的知识点中会有涉及。

三、额外讲解sync.Pool

我在项目中使用了sync.pool存储session对象，目的是为了保存和复用session这个临时对象，以减少内存分配，减低gc压力，那么sync.Pool是什么呢？以下是官方给出的解释（自己翻译的）：

• Pool是一个可以存取临时对象的集合。
• Pool中保存的item都可能在没有任何通知的情况下被自动释放掉，即如果Pool持有该对象的唯一引用，这个item就可能被回收。
• Pool在被多个线程使用的情况下是安全的。
• Pool的目的是缓存分配了但是未使用的item用于之后的重用，以减轻GC的压力。也就是说，pool让创建高效的并且线程安全的空闲列表更加容易，不过Pool并不适用于所有空闲列表。
• Pool的合理用法是用于管理一组被多个独立并发线程共享并可能重用的临时item。Pool提供了让多个线程分摊内存申请消耗的方法。
• Pool比较经典的一个例子在fmt包里，该Pool维护一个动态大小的临时输出缓存仓库，该仓库会在过载（许多线程活跃的打印时）增大，在沉寂时缩小。
• 另一方面，管理着短寿命对象的空闲列表不适合使用Pool，因为这种情况下内存申请消耗不能很好的分配。这时应该由这些对象自己实现空闲列表。

以下是Pool的数据类型：

type Pool struct {
    noCopy noCopy

    local     unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal
    localSize uintptr        // size of the local array

    // New optionally specifies a function to generate
    // a value when Get would otherwise return nil.
    // It may not be changed concurrently with calls to Get.
    New func() interface{}
}

// Local per-P Pool appendix.
type poolLocalInternal struct {
    private interface{}   // Can be used only by the respective P.
    shared  []interface{} // Can be used by any P.
    Mutex                 // Protects shared.
}

type poolLocal struct {
    poolLocalInternal

    // Prevents false sharing on widespread platforms with
    // 128 mod (cache line size) = 0 .
    pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

由注释我们也可以看出，其中的local成员的真实类型是一个poolLocal数组，而localSize是数组长度，poolLocal是真正保存数据的地方。private保存了一个临时对象，shared是保存临时对象的数组，而从private和shared的注释我们也可以看出，一个是属于特定的P私有的，一个是属于所有的P，至于这个P是什么，可以自行参考golang的调度模型，后期我也会专门写一篇相关的博客。其次，Pool是给每个线程分配了一个poolLocal对象，就是说local数组的长度，就是工作线程的数量(size := runtime.GOMAXPROCS(0))。当多线程在并发读写的时候，通常情况下都是在自己线程的poolLocal中存取数据，而只有当自己线程的poolLocal中没有数据时，才会尝试加锁去其他线程的poolLocal中“偷”数据。

我们可以看看Get函数，源码如下：

func (p *Pool) Get() interface{} {
    if race.Enabled {
        race.Disable()
    }
    l := p.pin()
    x := l.private
    l.private = nil
    runtime_procUnpin()
    if x == nil {
        l.Lock()
        last := len(l.shared) - 1
        if last >= 0 {
            x = l.shared[last]
            l.shared = l.shared[:last]
        }
        l.Unlock()
        if x == nil {
            x = p.getSlow()
        }
    }
    if race.Enabled {
        race.Enable()
        if x != nil {
            race.Acquire(poolRaceAddr(x))
        }
    }
    if x == nil && p.New != nil {
        x = p.New()
    }
    return x
}

这个函数的源码并不难读，在调用Get的时候首先会先在local数组中获取当前线程对应的poolLocal对象，然后再从poolLocal对象中获取private中的数据，如果private中有数据，则取出来直接返回。如果没有则先锁住shared，然后从shared中取出数据后直接返回，如果还是没有则调用getSlow函数。那么为什么这里要锁住shared呢？答案我们可以在getSlow中找到，因为当shared中没有数据的时候，会尝试去其他的poolLocal的shared中偷数据。

    // See the comment in pin regarding ordering of the loads.
    size := atomic.LoadUintptr(&p.localSize) // load-acquire
    local := p.local                         // load-consume
    // Try to steal one element from other procs.
    pid := runtime_procPin()
    runtime_procUnpin()
    for i := 0; i < int(size); i++ {
        l := indexLocal(local, (pid+i+1)%int(size))
        l.Lock()
        last := len(l.shared) - 1
        if last >= 0 {
            x = l.shared[last]
            l.shared = l.shared[:last]
            l.Unlock()
            break
        }
        l.Unlock()
    }
    return x

tip：该项目的源码（包含数据库的增删查改的demo）可以查看 源代码

有兴趣的可以关注我的个人公众号 ～

qrcode_for_gh_04e57fbebd02_258.jpg