美文网首页Redis
Redis之并发写入

Redis之并发写入

作者: ieasy_tm | 来源:发表于2017-08-28 19:14 被阅读0次

首先需要澄清一个事实：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~

相关文章

  • Redis之并发写入

    首先需要澄清一个事实：redis服务端是单线程处理客户端请求，也就是说客户端请求在服务端是串行化执行的，因此对服务...

  • redis常用操作及语句

    nosql系列之redis 特点： 1）并发写入 2）从海量数据查找某一条数据记录 3）高扩展...

  • Redis并发竞争key的解决方案详解

    架构师进阶之Redis专题系列，今天主要分享Redis高并发竞争解决方案。 01. 并发竞争的由来 1.Redis...

  • Leveldb之并发写入

    这篇文章想分析下Leveldb的写入流程以及Leveldb是如何处理并发写入问题的。阅读https://githu...

  • 实时数据并发写入 Redis 优化

    01 背景 当前架构的逻辑是将并发请求数据写入队列中，然后起一个单独的异步线程对数据进行串行处理。这种方式的好处就...

  • 无标题文章

    大并发写入案例： 高并发，大数据量写入：先把数据写入到内存，积累到一定的量后，在定时或者定量的写入到磁盘（减少磁盘...

  • redis

    redis Redis － 高并发，内存高速cache ，处理读写io的高并发容忍。 redis 通常被使用来将那...

  • 缓存架构之多级缓存（一）多级缓存及每一层的意义

    我们之前在缓存架构之redis中，主要讲解的是redis是如何支持高并发的，它的内部原理的基本思路等。在一个高并发...

  • redis高并发高可用

    redis 实现高并发主要依靠主从架构，一主多从. 对于性能来说，单主用来写入数据，单机几万QPS，多从用来查询数...

  • redis 的高并发高可用

    redis 实现高并发主要依靠主从架构，一主多从. 对于性能来说，单主用来写入数据，单机几万QPS，多从用来查询数...

网友评论

    本文标题：Redis之并发写入

    本文链接：https://www.haomeiwen.com/subject/jffidxtx.html

    延伸阅读

    深度阅读

    • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

    栏目导航

    热点阅读

    Redis

    关于我们|服务条款|联系我们|Redis之并发写入|投稿指南|网站地图|RSS订阅|排版工具|手机版

    提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

    Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

    备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

    本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

    所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！