小爬虫项目遇到的bug

项目背景是这样的， 三个请求数据的接口，是标准的 JWT 认证，提前五分钟刷新token，这个操作是在每次请求数据之前都会进行的检查。
业务场景是这样的： 每个账号都需要请求三个接口，为了简单就做成了定时任务的模式，所以每个账号下就会有三个任务，每次刷新完token，都会将token存入到数据库，以便于下次能重新拿到最新的token。
遇到的问题: 在某次查看日志的时候，发现每次刷新完token之后就会掉线。

伪代码

# -*- coding:utf-8  -*-
# @Time     : 2021-02-21 02:32
# @Author   : BGLB
# @Software : PyCharm


class Spider():

    def task_1():
        if not self.refresh_token():
            return False
            print('task_1....', self.token))
            time.sleep(5)

    def task_2():
        if not self.refresh_token():
            return False
        print('task_2...', self.token))
        time.sleep(5)

    def task_3():
        if not self.refresh_token():
            return False
        print('task_3....', self.token))
        time.sleep(5)

    def refresh_token(self, force: bool = False) -> bool:
        """
        force: 强制刷新token
        :return:
         bool
        """
        self.token = ''
        return True


def main():
    spider = Spider()
    task_1_thread = Thread(target=spider.task_1, )
    task_2_thread = Thread(target=spider.task_2, )
    task_3_thread = Thread(target=spider.task_3, )

    task_1_thread.start()
    task_2_thread.start()
    task_3_thread.start()
    while True:
        time.sleep(10)

if __name__ == '__main__':
    main()
    

找出背锅侠

定时任务 由于我们的每次启动任务的时候几乎每个账号的每个任务是同时启动的，在 task_1 任务中刷新token之后，新的token还没有保存到数据库，task_2 和 task_3 任务拿到的还是旧的token 所以还是需要去刷新token，这一刷新不就出大问题了。当然任务这块不是我写的，但是问题要解决啊，怀着对他人代码的敬畏之心，我毅然决然的一头扎入自己的代码之中…

解决之道

1. 找 chatgpt 寻一个分布式锁
2. 在每次刷新 token 前获取一个锁，如果没有拿到锁，就循环从redis中获取新的token, 删除这个新的token
3. 每次刷新完之后把最新的token放入到redis中以便于其他任务获取。

最终修改后的代码如下

# -*- coding:utf-8  -*-
# @Time     : 2023/6/8 0:25
# @Author   : BGLB
# @Software : PyCharm
import time
import uuid
from threading import Thread


def acquire_lock(redis_client, lock_name, acquire_time=10, time_out=10):
    """获取一个分布式锁"""
    identifier = str(uuid.uuid4())
    end = time.time()+acquire_time
    lock = "string:lock:"+lock_name
    while time.time()  end:
        if redis_client.setnx(lock, identifier):
            # 给锁设置超时时间, 防止进程崩溃导致其他进程无法获取锁
            redis_client.expire(lock, time_out)
            return identifier
        elif not redis_client.ttl(lock):
            redis_client.expire(lock, time_out)
        time.sleep(0.001)
    return False


def release_lock(redis_client, lock_name, identifier):
    """通用的锁释放函数"""
    lock = "string:lock:"+lock_name
    pip = redis_client.pipeline(True)
    while True:
        try:
            pip.watch(lock)
            lock_value = redis_client.get(lock)
            if not lock_value:
                return True

            if lock_value.decode() == identifier:
                pip.multi()
                pip.delete(lock)
                pip.execute()
                return True
            pip.unwatch()
            break
        except Exception:
            pass
    return False


def redis_client():
    pool = redis.ConnectionPool(**REDIS_CONFIG, decode_responses=True)
    r = redis.Redis(connection_pool=pool)
    return r


class Spider():

    def __init__(self, account):
        self.account = account
        self.token = ''
        self._redis_client = None

    def task_1(self):
        if not self.refresh_token():
            return False
        print('task_1....', self.token))
        time.sleep(5)

    def task_2(self):
        if not self.refresh_token():
            return False
        print('task_2...', self.token))
        time.sleep(5)

    def task_3(self):
        if not self.refresh_token():
            return False
        print('task_3....', self.token)
        time.sleep(5)

    def refresh_token(self, force: bool = False) -> bool:
        """
        force: 强制刷新token
        :return:
         bool
        """
        new_token = ''
        lock_key = f'refreshing_token_{self.account}'
        token_key = f'new_token_{self.account}'
        lock = acquire_lock(self.redis_client, lock_key, acquire_time=2)
        if not lock:
            # 从缓存中获取新的token
            time_start = time.time()
            while time.time() - time_start  5:
                new_token = self.redis_client.get(token_key)
                print('wait other task refreshing token')
                if new_token:
                    self.redis_client.delete(token_key)
                    break
                time.sleep(0.5)
        else:
            # 请求刷新token 的接口
            try:
                new_token = ''
            except Exception as e:
                raise e
            finally:
                release_lock(self.redis_client, lock_key, lock)

        self.token = new_token
        print('refresh_token finish')
        return True

    @property
    def redis_client(self):
        if not self._redis_client:
            self._redis_client = redis_client()
        return self._redis_client


def main():
    spider = Spider(account='123456')
    task_1_thread = Thread(target=spider.task_1, )
    task_2_thread = Thread(target=spider.task_2, )
    task_3_thread = Thread(target=spider.task_3, )

    task_1_thread.start()
    task_2_thread.start()
    task_3_thread.start()
    while True:
        time.sleep(10)


if __name__ == '__main__':
    main()

存疑

刚才在编写伪代码的过程中，我又发现一个问题，如果 task_1 在刷新token， task_1 和 task_2 都在等待中 , task_2 从reids中获取到了新的token，然后就会从redis删除这个token，这个时候呢， task_3会不会就拿不到刷新后的token了？万一这个 task_3 又比较重要，它的返回值决定了所有任务生死，那这个问题是不是就很严重了？
哎，说实话（听我给你编）这原本不是我代码的问题。这是调用的问题，是吧？原本这些检查不是我来处理的，是调用方去处理，每个模块有每个模块的功能啊 是不？还是把调用的代码改一下吧！

服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net