redis_lock_test.py

import time
import threading
import redis, redis_lock


def locking(conn):
    tid = str(threading.get_ident()).encode()
    lock = redis_lock.Lock(conn, 'loclock', id=tid, expire=10)
    if lock.acquire(blocking=False):
        print("Got the lock.", str(tid))
        # time.sleep(1)
        lock.reset()
    else:
        if lock.get_owner_id() == tid:
            print("The lock is already acquired in another thread.", str(tid))
        else:
            print("The lock is held in another thread.", str(tid))


def main(thread_count):
    threads = []
    for i in range(thread_count):
        conn = redis.Redis()
        t = threading.Thread(target=locking, args=(conn, ))
        threads.append(t)

    for t in threads:
        t.start()

    for t in threads:
        t.join()


if __name__ == '__main__':
    main(3)