高並發場景下數據重復插入的問題

高並發場景下，數據庫經常會發生數據重復插入的問題，這時候單單在插入前，查詢數據庫，判斷是否存在，再進行插入，往往不能保證數據唯一性。

查詢數據庫判斷是否存在
測試代碼： th_insert_test.py 每次插入前，去數據庫查詢，要插入的 User0-9 是否存在，若不存在則插入，若存在，則返回已經有。

#-*- coding:utf8 -*-

def db_op_thread_func(i, num_of_op):
    r = redis.Redis(host='127.0.0.1', port=6379, db=0)
    # r = redis.Redis(connection_pool=pool)

    conn = MySQLdb.connect(host="redisHost", port=3306, user="root", passwd="pass", db="blog")
    cursor = conn.cursor()

    for j in range(0, int(num_of_op)):
        nickname = 'User' + str(int(i % 10))
        lockkey = "lock"+nickname

        getsql = ("select ID from User  where Username = '%s'") % (nickname)
        cursor.execute(getsql)
        fetchData = cursor.fetchall()
        if not fetchData  : 
            sql = ("insert into User (Username)  values('%s')   ")%(nickname)
            cursor.execute(sql)
            id = int(conn.insert_id())
            print int(id)
            print "thread", i, ":", " num:", j
            conn.commit()
        else:
            print '已經有'

    conn.close()

if __name__ == "__main__":
        args = sys.argv
        num_of_thd = args[1]   ## 線程數
        num_of_op = args[2]    ## 每個線程的op數
        threads = []

        for i in range(0, int(num_of_thd)):
            threads.append(threading.Thread(target=db_op_thread_func, args=(i, num_of_op)))

        for t in threads:
            t.start()

        for t in threads:
            t.join()
、

　　

測試一下，運行 python th_insert_test.py 50 5 
50個線程，每個線程op數為5 

理想的運行結果： User0-9 ，10條數據。 
實際數據庫插入運行結果： 
結果1： 
 
結果2： 

可以看到 兩次分別產生56 和46 行，這樣在並發下是不可行的。

分布式鎖方案

基於 redis 的 setnx 來解決這一問題。

 

def db_op_thread_func(i, num_of_op):
    r = redis.Redis(host='redisHost', port=6379, db=0)
    conn = MySQLdb.connect(host="dbHost", port=3306, user="root", passwd="pass", db="blog")
    cursor = conn.cursor()

    for j in range(0, int(num_of_op)):
        nickname = 'User' + str(int(i % 10))
        lockkey = "lock"+nickname

        getsql = ("select ID from User  where Username = '%s'") % (nickname)
        cursor.execute(getsql)
        fetchData = cursor.fetchall()

        reply = r.setnx(lockkey, 1)
        if (reply == True):
            r.expire(lockkey, 30)
            RedisLock =  False
        else:
            RedisLock =  True

        if not fetchData  and RedisLock == False:
            sql = ("insert into User (Username)  values('%s')   ")%(nickname)
            cursor.execute(sql)
            id = int(conn.insert_id())
            print int(id)
            print "thread", i, ":", " num:", j
            conn.commit()
        else:
            print '已經有'
    conn.close()

　　

setnx key value 若 value 存在 則返回 False. 

運行測試： 
兩次插入，第一次插入10條，第二次插入0條。 

 

每當插入前設置 UserName 的一個 redis Lock ，expire 設置為30s ，這樣就可以利用 setnx 的原子性 來實現分布式鎖來保證數據唯一性。

點贊 1