如何在Python单元测试中高效启停Redis服务？解决端口占用问题

问题背景 #

• 单元测试中需多次启停Redis服务，但停止逻辑异常，再次启动时出现Address already in use错误
• 停止后执行netstat可见TIME_WAIT连接，怀疑是问题根源

错误日志 #

5638:M 01 Jul 2023 07:09:12.828 * monotonic clock: POSIX clock_gettime
5638:M 01 Jul 2023 07:09:12.828 # Warning: Could not create server TCP listening socket *:6379: bind: Address already in use
5638:M 01 Jul 2023 07:09:12.828 # Failed listening on port 6379 (TCP), aborting.
5632:signal-handler (1688195360) Received SIGTERM scheduling shutdown...
5632:M 01 Jul 2023 07:09:20.844 # User requested shutdown...
5632:M 01 Jul 2023 07:09:20.844 * Saving the final RDB snapshot before exiting.
5632:M 01 Jul 2023 07:09:20.849 * DB saved on disk
5632:M 01 Jul 2023 07:09:20.849 # Redis is now ready to exit, bye bye...

现有代码 #

class DBServer(object):
    def __init__(self):
        super().__init__()
        self.terminate = False
        self.__initialize_db()

    def __del__(self):
        self.terminate = True
        self.db_server.terminate()
        try:
            self.db_server.wait(timeout=10)
        except subprocess.TimeoutExpired as e:
            print("%s" % e)
        finally:
            self.db_server.kill()

    def __update_dns_entry(self):
        try:
            with open("/etc/hosts", "r+") as f:
                content = f.readlines()
                for l in content:
                    if "redis" in l:
                        return
                f.seek(0, 2)
                f.write("\n127.0.0.1\tredis")
        except:
            print("Exception occurred while trying to update /etc/hosts file %s" % traceback.format_exc())
            os.exit()

    def __initialize_db(self):
        self.__update_dns_entry()
        self.db_server = subprocess.Popen(
                ["/usr/bin/redis-server",], 
                close_fds=True)

    
    def start(self):
        time.sleep(5)
        self.add_db_entries()
        self.tid = threading.Thread(target=self.run)
        self.tid.start()

    def run(self):
        while not self.terminate:
            time.sleep(1)

    def stop(self):
        self.terminate = True
        redis_conn = redis.Redis(host = "redis", port = 6379, db=os.environ["DB"])
        redis_conn.flushall()
        self.tid.join()
        time.sleep(3)

问题分析与修复方案 #

核心问题点 #

1. stop方法未真正停止Redis进程：当前stop方法仅设置terminate标志、刷新数据并等待线程结束，但没有向Redis进程发送停止信号，Redis进程仍在运行，占用端口。
2. __del__方法不可靠：Python的垃圾回收时机不确定，__del__可能不会及时触发，导致Redis进程未被正确终止。
3. TIME_WAIT连接影响：即使Redis进程终止，之前的连接处于TIME_WAIT状态，默认情况下操作系统会保留该端口一段时间，导致新进程无法绑定。

修复步骤 #

1. 完善stop方法，主动停止Redis进程

修改stop方法，添加向Redis进程发送终止信号的逻辑：

def stop(self):
    self.terminate = True
    # 先关闭Redis连接并刷新数据
    try:
        redis_conn = redis.Redis(host="redis", port=6379, db=os.environ["DB"])
        redis_conn.flushall()
        redis_conn.shutdown()  # 发送SHUTDOWN命令优雅停止Redis
    except redis.exceptions.ConnectionError:
        # 处理Redis已断开的情况
        pass
    # 等待线程结束
    self.tid.join()
    # 确保Redis进程终止
    if hasattr(self, 'db_server') and self.db_server.poll() is None:
        self.db_server.terminate()
        try:
            self.db_server.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.db_server.kill()
    time.sleep(1)  # 缩短不必要的等待时间

2. 避免依赖__del__方法，显式调用stop

单元测试中，在每个测试用例结束后显式调用db_server.stop()，而不是依赖垃圾回收触发__del__。例如在测试框架的teardown方法中调用：

def teardown(self):
    if hasattr(self, 'db_server'):
        self.db_server.stop()

3. 配置Redis允许快速复用端口

启动Redis时添加参数，同时设置系统参数允许端口快速复用（仅在测试环境使用）：
修改__initialize_db方法：

def __initialize_db(self):
    self.__update_dns_entry()
    # 添加Redis启动参数，允许快速重启
    self.db_server = subprocess.Popen(
        ["/usr/bin/redis-server", "--port", "6379", "--tcp-keepalive", "0", "--bind", "0.0.0.0"],
        close_fds=True
    )
    # 设置系统参数，允许端口复用
    import subprocess
    try:
        subprocess.run(["sysctl", "-w", "net.ipv4.tcp_tw_reuse=1"], check=True)
        subprocess.run(["sysctl", "-w", "net.ipv4.tcp_tw_recycle=1"], check=True)
    except subprocess.CalledProcessError:
        print("Warning: Failed to set TCP reuse parameters")

4. 检查并清理残留进程

在启动Redis前，检查端口6379是否被占用，若有则杀死对应进程：

def __initialize_db(self):
    self.__update_dns_entry()
    # 检查并清理占用6379端口的进程
    import subprocess
    try:
        result = subprocess.run(
            ["lsof", "-t", "-i:6379"],
            capture_output=True, text=True, check=False
        )
        if result.stdout:
            pids = result.stdout.strip().split("\n")
            for pid in pids:
                subprocess.run(["kill", "-9", pid], check=True)
    except subprocess.CalledProcessError:
        print("Warning: Failed to kill existing Redis processes")
    # 启动Redis
    self.db_server = subprocess.Popen(
        ["/usr/bin/redis-server", "--port", "6379", "--tcp-keepalive", "0", "--bind", "0.0.0.0"],
        close_fds=True
    )

总结 #

通过显式控制Redis进程的启停、避免依赖不可靠的__del__方法、配置系统允许端口复用以及清理残留进程，可以解决多次启停Redis时出现的Address already in use问题。

内容的提问来源于stack exchange，提问作者Neeraj