影刀RPA进阶：我开发了一套店群管理系统，彻底解决100+店铺并发卡死痛点

22个窗口同时跑，内存稳在14GB，CPU 73%，一周零崩溃。不是电脑配置高，是调度层做对了。

去年冬天，一个做TikTok店群的老客户给我发来一段录屏。

屏幕上，影刀RPA同时开着18个浏览器窗口，CPU占用98%，鼠标已经卡成PPT。过了大概四十秒，整个画面冻住了。

他发了条语音：“林工，又死了。这已经是今天第三次了。我64GB内存的机器啊，怎么还不如我手动切号？”

我没急着回答。

远程上去看了一眼日志：18个窗口共用同一个user-data-dir，SQLite锁冲突导致大面积白屏，内存只升不降，最后系统直接杀了chrome进程。

这不是电脑的问题，是架构的问题。

影刀RPA本身没有“并发控制”，也没有“环境隔离”。你同时启动20个流程，它就给你开20个窗口，所有cookie、缓存全搅在一起。平台一看就知道是群控，你的店铺不封谁封？

那天晚上我开始写Alien。

三个月后，同样的机器，22个窗口并发跑了一周，串号0次，崩溃0次。

picture.image

picture.image 今天这篇文章，我把Alien的完整实现复盘出来。不讲虚的，全是代码、架构、以及那些让我通宵的坑。

一、店群并发卡死的真相：不是电脑不行，是影刀没有“底盘”

先说结论：影刀RPA是个优秀的“手脚”，但它没有“大脑”和“骨架”。

你让它跑流程，它就跑。但它不会管：

这个店铺的cookie和上一个店铺是不是混在一起了
同时开20个窗口内存会不会爆
某个窗口崩了要不要自动重启

更致命的是，影刀默认打开浏览器时，所有窗口共用一个user-data-dir。这意味着：

20个店铺的cookie、localStorage、缓存全在同一个目录里。

平台的检测脚本轻而易举就能发现：同一个IP段、同一个浏览器指纹、同一套缓存目录下，登录了几十个不同账号。这不是群控是什么？封。

而且Chrome的SQLite数据库（比如Cookies文件）不支持多进程高并发读写。当十几个窗口同时读写同一个数据库时，锁冲突频繁发生，轻则卡顿，重则整个窗口白屏崩溃。

所以我的方案很简单：

picture.image

把环境隔离和并发调度从影刀里拆出来，自己写一层。 影刀只负责一件事：通过Chrome的远程调试端口，在Alien已经开好的浏览器窗口里执行.flow流程。

二、Alien三层架构：各司其职，绝不越界

Alien的架构分为四层，从上到下：

UI层（PyQt6）：老板双击就能用，不用看命令行。
调度核心层（Python）：控制并发数、管理任务队列、回收资源。
环境隔离层（Chrome user-data-dir）：每个店铺独立的profile、指纹、代理。
执行层（影刀RPA）：只负责在指定窗口里点击、输入、读取。

用户只看到第1层。下面三层，全自动运行。

三、核心模块A：环境隔离矩阵——每个店铺都是一台“虚拟电脑”

3.1 UI设计：让老板一眼看懂“隔离”

第一版Alien是命令行。客户看了说：“这像黑客工具，我不敢用。”

我用PyQt6重写了界面。主界面三块：

左侧分组树：无限级嵌套。老板可以建“TK美区-服装”、“拼多多-食品”等分组，拖拽卡片就能移动店铺。
中间卡片墙：每个店铺一张卡片。卡片上只放老板最关心的信息：店铺名（大号字体）、代理IP国旗、Cookie剩余天数（绿/黄/红）、状态灯（🟢空闲/🔵执行中/🔴异常）。
右侧操作面板：点击卡片后显示详细配置。底部三个核心按钮——“批量导入”“手动打开”“批量校验”。

批量导入是真正的生产力工具。下载Excel模板，填上店铺名、平台、代理IP，点一下导入。Alien自动完成：

创建独立user-data-dir
注入随机指纹（分辨率、硬件并发数、WebGL、时区）
测试代理连通性
生成元数据

一百个店，三分钟。客户原话：“我以前招个人干两天，还总把代理填错。”

手动打开解决了老板的“亲眼看到才放心”心理。选中一个店铺，点一下，Alien用该店铺的独立profile启动Chrome窗口。老板可以亲自进去检查后台，用完直接关掉，不影响自动化。

3.2 技术拆解：一个店铺 = 一个独立Profile + 随机指纹

每个店铺的profile存放在./alien_profiles/{env_id}。创建时从空白模板拷贝，然后注入指纹。

import shutil
import json
import random
from pathlib import Path

class EnvFactory:
    BLANK = Path("./assets/blank_profile")
    ROOT = Path("./alien_profiles")

    @classmethod
    def create(cls, env_id: str, proxy: str = None) -> Path:
        target = cls.ROOT / env_id
        if target.exists():
            bak = target.with_suffix(".bak")
            shutil.move(str(target), str(bak))
            shutil.rmtree(str(bak), ignore_errors=True)

        shutil.copytree(cls.BLANK, target)
        cls._inject_fingerprint(target)

        if proxy:
            (target / "proxy.txt").write_text(proxy)

        return target

    @classmethod
    def _inject_fingerprint(cls, profile_path: Path):
        prefs_file = profile_path / "Default" / "Preferences"
        if not prefs_file.exists():
            return

        with open(prefs_file, 'r', encoding='utf-8') as f:
            prefs = json.load(f)

        # 屏幕分辨率随机（16:9常见值）
        widths = [1366, 1440, 1536, 1600, 1920, 2560]
        w = random.choice(widths)
        h = int(w * 0.5625) + random.randint(-20, 20)
        prefs.setdefault("web_prefs", {})["screen"] = {"width": w, "height": h}

        # 硬件并发数随机
        prefs["hardware_concurrency"] = random.choice([2, 4, 6, 8, 12])

        # WebGL渲染器随机
        renderers = [
            "ANGLE (NVIDIA, NVIDIA GeForce GTX 1060)",
            "ANGLE (Intel, Intel(R) UHD Graphics 620)",
            "Google SwiftShader",
            "ANGLE (AMD, Radeon RX 580)"
        ]
        prefs.setdefault("webgl", {})["renderer"] = random.choice(renderers)

        # 时区随机
        timezones = ["America/New_York", "America/Los_Angeles", "Europe/London", "Asia/Tokyo"]
        prefs["timezone"] = random.choice(timezones)

        with open(prefs_file, 'w', encoding='utf-8') as f:
            json.dump(prefs, f, indent=2)

启动浏览器，暴露远程调试端口供影刀连接：

import subprocess
import socket
import time

def launch_chrome(env_id: str, debug_port: int):
    profile_dir = Path(f"./alien_profiles/{env_id}")
    proxy_file = profile_dir / "proxy.txt"
    proxy = proxy_file.read_text().strip() if proxy_file.exists() else None

    chrome_path = "C:/Program Files/Google/Chrome/Application/chrome.exe"
    if not Path(chrome_path).exists():
        chrome_path = "./assets/chrome_portable/chrome.exe"

    cmd = [
        chrome_path,
        f"--user-data-dir={profile_dir}",
        f"--remote-debugging-port={debug_port}",
        "--no-first-run",
        "--disable-blink-features=AutomationControlled",  # 关键：隐藏自动化痕迹
        "--disable-features=TranslateUI,PasswordImport",
        "--disable-background-networking",
        "--disable-sync",
    ]
    if proxy:
        cmd.append(f"--proxy-server={proxy}")

    proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                            creationflags=subprocess.CREATE_NO_WINDOW)

    for _ in range(30):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        if sock.connect_ex(('127.0.0.1', debug_port)) == 0:
            sock.close()
            return proc
        sock.close()
        time.sleep(0.3)
    raise TimeoutError(f"端口{debug_port}未响应")

3.3 踩坑：Chrome把我的指纹覆盖了

上线第三天，客户说所有店铺指纹一模一样——都是1920x1080。

远程上去，打开Preferences文件，发现我写的指纹配置全没了，被Chrome恢复成默认值。

查了一夜才知道：Chrome启动时会“校验”Preferences。如果某些字段格式不对或值超出合理范围，就判定文件损坏，直接覆盖。

解决方案：只修改Chrome官方文档明确允许的字段，并且通过CDP在启动后二次注入剩余属性。修好后指纹再也没有被重置过。

四、核心模块B：调度编排——22个窗口并发不崩的秘密

环境隔离好了，下一个问题：怎么让100个店铺的任务有序执行？

你不能同时开100个窗口——内存会爆。也不能串行跑——100个店×5分钟=8小时。

我用自己的工控机（32GB内存，8核i7）做了压测：

并发窗口	CPU	内存	100任务耗时	稳定性
10	30%	6GB	50分钟	100%
15	45%	9GB	34分钟	100%
20	62%	13GB	26分钟	98%
22	73%	15GB	24分钟	96%
25	88%	18GB	23分钟	72%

22是甜点值。比25只慢1分钟，但稳定性从72%飙升到96%。Alien默认并发上限就是22。

4.2 调度器核心：信号量 + 环境锁 + 优先级队列

import threading
import queue
from dataclasses import dataclass, field

@dataclass(order=True)
class Task:
    priority: int = 5   # 1最高，10最低
    task_id: str = field(compare=False)
    env_id: str = field(compare=False)
    flow_file: str = field(compare=False)
    retries: int = field(default=3, compare=False)

class AlienScheduler:
    def __init__(self, max_concurrent=22):
        self.sem = threading.Semaphore(max_concurrent)
        self.q = queue.PriorityQueue()
        self.env_busy = {}      # env_id -> bool
        self.lock = threading.Lock()
        self._start_workers(max_concurrent)

    def _start_workers(self, n):
        for _ in range(n):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()

    def submit(self, task: Task):
        self.q.put((task.priority, task))

    def _worker(self):
        while True:
            prio, task = self.q.get()
            with self.lock:
                if self.env_busy.get(task.env_id, False):
                    # 该环境正忙，重新入队并降优先级
                    self.q.put((prio + 1, task))
                    continue
                self.env_busy[task.env_id] = True

            self.sem.acquire()
            try:
                self._execute(task)
            finally:
                self.sem.release()
                with self.lock:
                    self.env_busy[task.env_id] = False
            self.q.task_done()

    def _execute(self, task: Task):
        import requests
        try:
            resp = requests.post("http://127.0.0.1:18888/run_flow",
                                 json={"flow": task.flow_file, "env_id": task.env_id},
                                 timeout=300)
            if resp.status_code != 200 and task.retries > 0:
                task.retries -= 1
                self.q.put((task.priority + 1, task))
        except Exception:
            if task.retries > 0:
                task.retries -= 1
                self.q.put((task.priority + 1, task))

关键设计：

信号量：全局最多22个任务同时执行。
环境锁（env_busy）：同一个店铺不会同时跑两个任务，防止状态冲突。
优先级队列：紧急任务（如订单发货）可以插队。
自动重试：失败后降级优先级重试，避免死循环。

4.3 编排流：拖拽连线，业务傻瓜化

老板不想每天手动选流程。Alien的“编排流”编辑器允许预设完整业务流水线。

左侧组件库：启动环境、执行影刀流程、等待、条件判断、循环、发送钉钉。中间画布拖拽连线。

一个真实的TikTok自动上货+报活动流：

循环遍历“美区主店”分组 2. 执行export_products.flow（导出商品）
1. 运行本地Python脚本压缩图片
2. 执行batch_publish.flow（批量发布）
3. 如果发布成功数>10，执行apply_activity.flow（报名活动）
循环结束
发送钉钉汇总

保存后，老板每天早上点一下“运行编排流”，然后去喝茶。原来两个人盯半天的工作，现在20分钟跑完。

五、底层封装：为什么客户觉得“这软件是大厂出的”

5.1 PyQt6：从黑框框到专业感

命令行版本被客户嫌弃后，我用PyQt6重写了界面。深灰主题+蓝绿强调色，所有字体14px以上，每个操作都有进度条。系统托盘支持最小化后台运行。

客户第二次看到界面时说：“这个正规多了。”

5.2 Nuitka打包：双击即用

Python打包一直是痛点。PyInstaller打出来的exe经常被杀毒软件误报，而且体积大。

Nuitka把Python编译成C++再链接，生成的exe几乎不误报，启动速度快。

打包命令：

nuitka --standalone --onefile --windows-console-mode=disable --enable-plugin=pyqt6 --include-data-dir=./assets=assets --include-data-dir=./blank_profile=blank_profile --windows-icon-from-ico=alien.ico main.py

客户拿到Alien.exe，双击，登录，使用。不需要装Python，不需要配Chrome（内置便携版）。零配置。

5.3 安全验证：防止白嫖

做了三层简单防护：

机器码绑定（CPU+主板+MAC生成指纹，RSA签名license）。
核心模块用Cython编译成.pyd，反编译难度大。
可选联网心跳验证。

拦不住顶级黑客，但挡住了99%的复制粘贴。

六、血泪教训：线上永远不会按剧本走

炸弹1：内存泄漏，跑两天崩了

排查发现任务结束后Chrome进程已杀，但Python里还保留着Popen对象引用。修复：在finally里del proc并调用gc.collect()。

炸弹2：代理切换时新环境用了旧IP

新进程启动时发现旧Chrome进程还没退出，复用了它的网络配置。修复：启动前强制杀死所有使用同一profile的Chrome进程，等待500ms再启动。

炸弹3：Windows更新后Chrome路径变了

代码里写死了Program Files，更新后变成了Program Files (x86)。修复：动态检测注册表和常见目录，找不到就用内置便携版。

每一个炸弹都让我通宵，但也正是这些坑，让Alien从“能跑”变成了“稳跑”。

七、结语

Alien不是黑科技。它就是Python、Chrome命令行参数、影刀RPA的合理组合。

但它解决了一个真实到让人头秃的问题：多店铺高并发自动化时的环境隔离和资源调度。

现在这套系统跑在十几个工作室的电脑上，日均处理店铺数超过2000。累计节省的人力成本超过五十万。

如果你也在被串号和并发卡死折磨，希望这篇文章能给你一个清晰的解决路径。

我是林焱，一个还在写代码、接电话、半夜查日志的独立开发者。

（全文完）