高并发场景下数据重复插入的问题

绝地灬酷狼 2022-05-25 05:07 247阅读 0赞

高并发场景下，数据库经常会发生数据重复插入的问题，这时候单单在插入前，查询数据库，判断是否存在，再进行插入，往往不能保证数据唯一性。

### 查询数据库判断是否存在 ###

测试代码： `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  
![这里写图片描述][70]

理想的运行结果： User0-9 ，10条数据。  
实际数据库插入运行结果：  
结果1：  
![这里写图片描述][70 1]  
结果2：  
![这里写图片描述][70 2]

可以看到 两次分别产生56 和46 行，这样在并发下是不可行的。

### 分布式锁方案 ###

基于 `redis` 的 `setnx` 来解决这一问题。

Python 代码：

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`.  
![这里写图片描述][70 3]

运行测试：  
两次插入，第一次插入10条，第二次插入0条。  
![这里写图片描述][70 4]

每当插入前设置 `UserName` 的一个 `redis Lock` ，`expire` 设置为30s ，这样就可以利用 `setnx` 的原子性 来实现分布式锁来保证数据唯一性。

[70]: /images/20220525/28bc10f72f994b28b24a4a3bf8e14d80.png
[70 1]: /images/20220525/e29533344b274062979ec48f6ad62102.png
[70 2]: /images/20220525/db5285fa605648e2b3076bc01cd4e4ff.png
[70 3]: /images/20220525/44233c40ed094725a0a78b49daa4e656.png
[70 4]: /images/20220525/99b75502965b4ea1a34c8a68631cb006.png