Redis之并发写入

首先需要澄清一个事实：redis服务端是单线程处理客户端请求，也就是说客户端请求在服务端是串行化执行的，因此对服务端来说，并不存在并发问题。但业务方却存在并发操作redis中的同一个key的情况。所以如何让A客户端知道B客户端正在操作它想操作的 key，就成了必须要讨论的问题。

那么开始总结下方案吧：

1. SETNX key value      //key存在就不做任何操作，返回0；不存在操作成功返回1

这种方式通过对需要操作的key加锁来保证并发操作的串行化。这里我们以Golang代码为例来举例说明该操作。先看多个协程写同一个key的情况。代码如下：

package main
import (
    "fmt"
    "github.com/garyburd/redigo/redis"
    "runtime"
    "sync"
    "time"
)
var w sync.WaitGroup
func newRdsPool(server, auth string) *redis.Pool {
    return &redis.Pool{
        MaxIdle:     100,
        MaxActive:   30,
        IdleTimeout: 60 * time.Second,
        Dial: func() (redis.Conn, error) {
            c, err := redis.Dial("tcp", server)
            if err != nil {
                return nil, err
            }
            if auth == "" {
                return c, err
            }
            if _, err := c.Do("AUTH", auth); err != nil {
                c.Close()
                return nil, err
            }
            return c, err
        },
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            _, err := c.Do("PING")
            return err
        },
    }
}
func g1(r redis.Conn) {
    for i := 0; i < 2; i++ {
        if _, err := redis.String(r.Do("set", "hello", "1")); err != nil {
            fmt.Println(err)
        }
        time.Sleep(10 * time.Millisecond)
    }
    w.Done()
}
func g2(r redis.Conn) {
    for i := 0; i < 2; i++ {
        if _, err := redis.String(r.Do("set", "hello", "2")); err != nil {
            fmt.Println(err)
        }
        time.Sleep(10 * time.Millisecond)
    }
    w.Done()
}
func main() {
    w.Add(2)
    runtime.GOMAXPROCS(runtime.NumCPU())
    var rc1 redis.Conn = newRdsPool(`127.0.0.1:6379`, ``).Get()
    var rc2 redis.Conn = newRdsPool(`127.0.0.1:6379`, ``).Get()
    defer rc1.Close()
    defer rc2.Close()
    go g1(rc1)
    go g2(rc2)
    w.Wait()
}

执行上面的代码之后，hello的值在1和2之间徘徊。希望出现的是如果协程1在操作时候，协程2就放弃操作，也即让操作串行化。这样就需要有一个锁来保证不能同时让两个协程进去临界区。setnx = set if not exists 不存在返回1，存在返回0。通过这个机制可以判断当前的lock是否已经被设置了。lock必须给一个过期时间，因为很有可能goroutine1在do work的时候出现panic，这样就导致goroutine2一直在尝试获取锁。

package main

import (
    "fmt"
    "github.com/garyburd/redigo/redis"
    "runtime"
    "sync"
    "time"
)

var w sync.WaitGroup

func newRdsPool(server, auth string) *redis.Pool {
    return &redis.Pool{
        MaxIdle:     100,
        MaxActive:   30,
        IdleTimeout: 60 * time.Second,
        Dial: func() (redis.Conn, error) {
            c, err := redis.Dial("tcp", server)
            if err != nil {
                return nil, err
            }
            if auth == "" {
                return c, err
            }
            if _, err := c.Do("AUTH", auth); err != nil {
                c.Close()
                return nil, err
            }
            return c, err
        },
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            _, err := c.Do("PING")
            return err
        },
    }
}

func g1(r redis.Conn) {
    var lock int64
    var lock_timeout int64 = 2
    var lock_time int64
    var now int64
    for lock != 1 {
        now = time.Now().Unix()
        lock_time = now + lock_timeout
        lock, err1 := redis.Int64(r.Do("setnx", "foo", lock_time))
        lockValue1, err2 := redis.Int64(r.Do("get", "foo"))
        if lock == 1 && err1 == nil {
            break
        } else {
            if now > lockValue1 && err2 == nil {
                lockValue2, err3 := redis.Int64(r.Do("getset", "foo", lock_time))
                if err3 == nil && now > lockValue2 {
                    break
                } else {
                    fmt.Println(`g1 not get lock`)
                    time.Sleep(1000 * time.Millisecond)
                }

            } else {
                fmt.Println(`g1 not get lock`)
                time.Sleep(1000 * time.Millisecond)
            }
        }
    }
    for i := 0; i < 5; i++ {
        if _, err := redis.String(r.Do("set", "hello", "1")); err != nil {
            fmt.Println(err)
        }
        fmt.Println(`g1 now work... `)
        time.Sleep(1 * time.Second)
    }
    if time.Now().Unix() < lock_time {
        if _, err4 := redis.Int64(r.Do("del", "foo")); err4 != nil {
            fmt.Println(err4)
        }
    }
    w.Done()
}

func g2(r redis.Conn) {
    var lock int64
    var lock_timeout int64 = 2
    var lock_time int64
    var now int64
    for lock != 1 {
        now = time.Now().Unix()
        lock_time = now + lock_timeout
        lock, err1 := redis.Int64(r.Do("setnx", "foo", lock_time))
        lockValue1, err2 := redis.Int64(r.Do("get", "foo"))
        if lock == 1 && err1 == nil {
            break
        } else {
            if now > lockValue1 && err2 == nil {
                lockValue2, err3 := redis.Int64(r.Do("getset", "foo", lock_time))
                if err3 == nil && now > lockValue2 {
                    break
                } else {
                    fmt.Println(`g2 not get lock`)
                    time.Sleep(1000 * time.Millisecond)
                }
            } else {
                fmt.Println(`g2 not get lock`)
                time.Sleep(1000 * time.Millisecond)
            }
        }
    }
    for i := 0; i < 5; i++ {
        if _, err := redis.String(r.Do("set", "hello", "2")); err != nil {
            fmt.Println(err)
        }
        fmt.Println(`g2 now work... `)
        time.Sleep(1 * time.Second)
    }
    if time.Now().Unix() < lock_time {
        if _, err4 := redis.Int64(r.Do("del", "foo")); err4 != nil {
            fmt.Println(err4)
        }
    }
    w.Done()
}

func main() {
    w.Add(2)
    runtime.GOMAXPROCS(runtime.NumCPU())
    var rc1 redis.Conn = newRdsPool(`127.0.0.1:6379`, ``).Get()
    var rc2 redis.Conn = newRdsPool(`127.0.0.1:6379`, ``).Get()
    defer rc1.Close()
    defer rc2.Close()
    go g1(rc1)
    go g2(rc2)
    w.Wait()
}

上面的代码给出了两个goroutine通过锁达到串行化操作同一个key的效果。好了，setnx的解法就到此为止了。

2.  MULTI、DISCARD、 EXEC  //  redis事务

利用MULTI 、DISCARD、 EXEC可以完成上面类似的操作，为了证明两个goroutine的执行是串行化的，在goroutine1的最后删除了hello这个key，goroutine2最后的输出来没有获取到任何东西nil。这也证明了MULTI和EXEC之间的逻辑是一个执行体，不受外界的干扰，但这里要说明下MULTI和EXEC只能保证它们之间的执行是一个整体不受干扰，但如果在A执行事务之前就有人把hello的值修改了，这样A事务就会在修改后的结果上继续操作。这种情况是否正常需要就具体业务来分析。针对这种情况这里不做过多分析了。

package main
import (
    "fmt"
    "github.com/garyburd/redigo/redis"
    "runtime"
    "sync"
    "time"
)

var w sync.WaitGroup

func newRdsPool(server, auth string) *redis.Pool {
    return &redis.Pool{
        MaxIdle:     100,
        MaxActive:   30,
        IdleTimeout: 60 * time.Second,
        Dial: func() (redis.Conn, error) {
            c, err := redis.Dial("tcp", server)
            if err != nil {
                return nil, err
            }
            if auth == "" {
                return c, err
            }
            if _, err := c.Do("AUTH", auth); err != nil {
                c.Close()
                return nil, err
            }
            return c, err
        },
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            _, err := c.Do("PING")
            return err
        },
    }
}

func g1(r redis.Conn) {
    r.Send("MULTI")
    fmt.Println(`g1 MULTI begin ...`)
    for i := 0; i < 5; i++ {
        if _, err := redis.String(r.Do("set", "hello", "10")); err != nil {
            fmt.Println(err)
        }
        fmt.Println(`g1 now work... `)
    }
    time.Sleep(1 * time.Second)
    r.Send("DEL", "hello")
    g, _ := r.Do("EXEC")
    fmt.Println(`g1 MULTI end ...`, g)
    w.Done()
}

func g2(r redis.Conn) {
    r.Send("MULTI")
    fmt.Println(`g2 MULTI begin ...`)
    for i := 0; i < 5; i++ {
        if _, err := redis.String(r.Do("get", "hello")); err != nil {
            fmt.Println(err)
        }
        fmt.Println(`g2 now work... `)
    }
    time.Sleep(1 * time.Second)
    g, _ := r.Do("EXEC")
    fmt.Println(`g2 MULTI end ...`, g)
    w.Done()
}

func main() {
    w.Add(2)
    runtime.GOMAXPROCS(runtime.NumCPU())
    var rc1 redis.Conn = newRdsPool(`127.0.0.1:6379`, ``).Get()
    var rc2 redis.Conn = newRdsPool(`127.0.0.1:6379`, ``).Get()
    defer rc1.Close()
    defer rc2.Close()
    go g1(rc1)
    go g2(rc2)
    w.Wait()
}

如果你有更多优雅的方式解决这类问题，请在评论区拍砖. end~

最后编辑于：2017.12.10 04:23:48

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 221,273评论 6赞 515
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 94,349评论 3赞 398
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 167,709评论 0赞 360
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 59,520评论 1赞 296
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 68,515评论 6赞 397
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 52,158评论 1赞 308
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,755评论 3赞 421
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,660评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 46,203评论 1赞 319
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,287评论 3赞 340
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,427评论 1赞 352
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 36,122评论 5赞 349
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,801评论 3赞 333
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,272评论 0赞 23
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,393评论 1赞 272
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 48,808评论 3赞 376
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,440评论 2赞 359

Redis之并发写入

推荐阅读更多精彩内容