threads 线程与同步

大约 12 分钟

threads 线程与同步

threads 是 ScriptX 里负责“开工作线程、在线程上挂定时器、做同步控制、在线程间传结果”的全局对象。

它和普通网页里的 Promise、setTimeout 不是一个层级的概念。这里的 threads.start(...) 真会拉起一个独立的脚本工作线程，而且那个线程自己还能再挂 setTimeout、setInterval、setImmediate。

这页会把下面这些最容易让人混的地方都讲清楚：

threads.start(...) 返回的到底是什么对象。
全局定时器和线程对象上的定时器分别挂在哪个线程上。
threads.exit() 在主线程和工作线程里行为为什么不一样。
disposable、atomic、lock、condition 各适合解决什么问题。
sync(func, lock?) 和 Java 锁对象之间是什么关系。

先记住这 10 条

threads 是全局对象，直接写 threads.start(...) 就能用。
threads.start(fn) 会立即启动一个单线程工作线程，返回一个线程句柄对象。
全局 setTimeout / setInterval / setImmediate 会挂到“当前所在的脚本线程”上，不一定总在主线程。
thread.setTimeout(...) 这类方法是显式把任务挂到某个线程句柄上。
定时器回调现在支持额外参数，setTimeout(fn, 100, a, b) 这种写法是有效的。
threads.exit() 如果在工作线程里调用，只退出当前工作线程；如果在主线程里调用，会直接中断整段脚本。
threads.disposable() 最适合做“线程 A 算完结果，线程 B 等待接收”。
threads.atomic() 返回的是 Java 的 AtomicLong，适合简单计数，不适合存复杂对象。
threads.lock() 返回的是 Java 的 ReentrantLock，newCondition() 再往下就是 Java 原生同步语义了。
sync(func, lock?) 是最省事的同步入口，不想手写 lock.lock() / unlock() 时优先用它。

`threads.start(action)`

启动一个新的脚本工作线程。

参数

参数	类型	可填值	说明
`action`	`function`	无参或自定义闭包函数	线程入口函数，必填

返回值

thread

这里返回的不是 Java 原生 Thread，而是 ScriptX 暴露出来的线程句柄对象。后面这组方法都挂在它上面：

thread.waitFor()
thread.join(...)
thread.interrupt()
thread.isAlive()
thread.setTimeout(...)
thread.setInterval(...)
thread.setImmediate(...)

说明

每次调用都会创建一个新的单线程调度器。
线程名字形如 Spawn-1、Spawn-2。
线程入口函数会立即开始跑，不需要你再手动 start()。
线程入口函数结束后，如果它自己挂的异步任务也都执行完了，这个线程才算真正结束。

示例

const worker = threads.start(function () {
  log("worker started");
  sleep(300);
  log("worker finished");
});

worker.join();
log("main done");

`threads.shutDownAll()`

中断当前还活着的所有工作线程。

参数

无。

返回值

null

说明

只处理通过 threads.start(...) 拉起来的工作线程。
不会把“主脚本线程”一起关掉。
每个线程上已经挂的 setTimeout / setInterval / setImmediate 也会一起取消。

示例

threads.start(function () {
  while (true) {
    sleep(1000);
  }
});

sleep(200);
threads.shutDownAll();

`threads.currentThread()`

拿到“当前代码正在运行的脚本线程句柄”。

参数

无。

返回值

thread

说明

如果你现在在主脚本里调用，拿到的是主线程句柄。
如果你现在在 threads.start(...) 的工作线程里调用，拿到的是那个工作线程句柄。
这个方法很适合写“我不关心自己现在在哪个线程里，但我想把后续任务继续挂在当前线程上”。

示例

const self = threads.currentThread();
self.setImmediate(function () {
  log("still on current thread");
});

`threads.getMainThread()`

拿到主脚本线程句柄。

参数

无。

返回值

thread

说明

不管你现在在哪个工作线程里，这个方法都会返回主线程句柄。
适合你想“从工作线程回到主脚本线程调度一个回调”时使用。

示例

const mainThread = threads.getMainThread();

threads.start(function () {
  mainThread.setImmediate(function () {
    log("back on main thread");
  });
});

`threads.allThreads()`

列出当前还活着的所有工作线程。

参数

无。

返回值

thread[]

说明

只返回还活着的工作线程。
已经执行完、已经被中断的线程不会出现在结果里。
主线程不会出现在这个数组里。

示例

const list = threads.allThreads();
log(`running=${list.length}`);

`threads.hasRunningThreads()`

判断当前是否还有工作线程没结束。

参数

无。

返回值

boolean

示例

if (threads.hasRunningThreads()) {
  log("还有后台线程没跑完");
}

`threads.exit()`

退出当前线程，或者在主线程里直接中断整段脚本。

参数

无。

返回值

工作线程里返回 null；主线程里会直接中断脚本，后面的代码不会继续执行。

说明

在工作线程里：只会中断当前工作线程。
在主线程里：等价于中断整段脚本运行时。
这也是为什么同一个 threads.exit()，你会觉得有时“只是停掉一个线程”，有时“整段脚本都没了”。

示例

threads.start(function () {
  log("before exit");
  threads.exit();
  log("这行不会再执行");
});

`threads.disposable()`

创建一个“可阻塞等待的单值容器”。

参数

无。

返回值

disposable

什么时候适合用

工作线程算完结果，主线程想阻塞等结果。
某个线程要“等另一个线程发信号”。
你只需要传 1 份结果，不想上锁、不想自己写条件变量。

示例

const box = threads.disposable();

threads.start(function () {
  sleep(500);
  box.setAndNotify({ ok: true, value: 123 });
});

const result = box.blockedGet(1000);
log(JSON.stringify(result));

`threads.atomic(initial?)`

创建一个原子整数对象。

参数

参数	类型	可填值	默认值	说明
`initial`	`number`	任意数字	`0`	初始值

返回值

atomic

说明

当前返回的是 Java AtomicLong。
适合做计数器、累计次数、跨线程自增。
如果你传的不是数字，会回退到 0。

示例

const counter = threads.atomic(0);

threads.start(function () {
  counter.incrementAndGet();
});

threads.start(function () {
  counter.incrementAndGet();
});

`threads.lock()`

创建一个可重入锁。

参数

无。

返回值

lock

说明

当前返回的是 Java ReentrantLock。
如果你需要更细粒度的同步控制，可以配合 newCondition() 用。
如果你只是想把一个函数变成“同一时刻只允许一个线程进来”，优先考虑 sync(func, lock?)，写起来更省事。

示例

const lock = threads.lock();

lock.lock();
try {
  log("critical section");
} finally {
  lock.unlock();
}

`thread.waitFor()`

等待线程真正启动完成。

参数

无。

返回值

null

说明

这个方法主要防止你在线程还没进入可调度状态时，就急着往上挂任务。
对主线程句柄来说，它是空操作，不会阻塞。

示例

const worker = threads.start(function () {
  log("ready");
});

worker.waitFor();
worker.setImmediate(function () {
  log("after started");
});

`thread.join(timeoutMs?)`

等待线程结束。

参数

参数	类型	可填值	默认值	说明
`timeoutMs`	`number`	任意数字	`0`	等待超时，`<= 0` 表示一直等

返回值

null

说明

timeoutMs <= 0 时会一直等到线程结束。
大于 0 时最多等这么久。
对主线程句柄来说，它是空操作。

示例

const worker = threads.start(function () {
  sleep(800);
});

worker.join(300); // 最多等 300ms
log(worker.isAlive());

`thread.interrupt()`

中断线程，并取消它挂着的定时任务。

参数

无。

返回值

null

说明

对工作线程：会停止接收新任务、取消已挂定时器、打断线程。
对主线程句柄：效果等同于中断整段脚本。

示例

const worker = threads.start(function () {
  while (true) {
    sleep(1000);
  }
});

sleep(300);
worker.interrupt();

`thread.isAlive()`

判断线程当前是否还活着。

参数

无。

返回值

boolean

说明

工作线程：已经启动且还没结束时返回 true。
主线程句柄：脚本运行时还活着就返回 true。

示例

const worker = threads.start(function () {
  sleep(500);
});

log(worker.isAlive());
worker.join();
log(worker.isAlive());

`thread.setTimeout(callback, delayMs?, ...args)`

在指定线程上挂一个单次定时器。

参数

参数	类型	可填值	默认值	说明
`callback`	`function`	回调函数	必填	不传会抛错
`delayMs`	`number`	任意数字	`0`	延迟毫秒数
`...args`	`any`	任意值	无	额外传给回调的参数

返回值

number

说明

回调会在这个 thread 对应的线程里执行。
如果线程还没真正启动，会抛 thread has not started yet。
如果线程已经结束，会抛 thread has already finished。
delayMs 负数会按 0 处理。

示例

const worker = threads.start(function () {});
worker.waitFor();

worker.setTimeout(function (name, count) {
  log(name, count);
}, 200, "job", 3);

`thread.clearTimeout(id)`

取消指定线程上的单次定时器。

参数

参数	类型	说明
`id`	`number`	任务编号

返回值

boolean

示例

const id = threads.currentThread().setTimeout(function () {
  log("never");
}, 1000);

threads.currentThread().clearTimeout(id);

`thread.setInterval(callback, delayMs?, ...args)`

在指定线程上挂一个循环定时器。

参数

和 thread.setTimeout(...) 一样。

返回值

number

说明

每次回调执行完，如果线程仍然活着，就会继续调下一轮。
如果回调内部抛出异常，当前循环会停止。

示例

const worker = threads.start(function () {});
worker.waitFor();

let n = 0;
const id = worker.setInterval(function () {
  n += 1;
  log(`tick=${n}`);
}, 300);

`thread.clearInterval(id)`

取消指定线程上的循环定时器。

参数

参数	类型	说明
`id`	`number`	任务编号

返回值

boolean

示例

const worker = threads.start(function () {});
worker.waitFor();

const id = worker.setInterval(function () {
  log("tick");
}, 1000);

worker.clearInterval(id);

`thread.setImmediate(callback, ...args)`

在指定线程上尽快执行一个回调。

参数

参数	类型	可填值	说明
`callback`	`function`	回调函数	必填
`...args`	`any`	任意值	额外传给回调的参数

返回值

number

说明

当前实现本质上等价于“这个线程上的 setTimeout(callback, 0, ...args)”。
它不是微任务，也不是 Promise 队列概念。
更适合表达“别阻塞当前这段逻辑，等线程调度一下再做”。

示例

const worker = threads.start(function () {});
worker.waitFor();

worker.setImmediate(function (text) {
  log(text);
}, "run soon");

`thread.clearImmediate(id)`

取消线程上的 setImmediate(...) 任务。

参数

参数	类型	说明
`id`	`number`	任务编号

返回值

boolean

示例

const worker = threads.start(function () {});
worker.waitFor();

const id = worker.setImmediate(function () {
  log("never");
});

worker.clearImmediate(id);

`setImmediate(callback, ...args)`

全局即时调度函数。

参数

和 thread.setImmediate(...) 一样。

返回值

number

说明

它会挂到“当前正在执行这段代码的脚本线程”上。
如果你当前就在工作线程里，那它不是回主线程，而是仍然挂在当前工作线程上。
如果你就是想挂到某个明确线程，优先写 thread.setImmediate(...)，更直观。

示例

setImmediate(function (a, b) {
  log(a + b);
}, 1, 2);

`clearImmediate(id)`

取消全局 setImmediate(...) 任务。

参数

参数	类型	说明
`id`	`number`	任务编号

返回值

boolean

示例

const id = setImmediate(function () {
  log("never");
});

clearImmediate(id);

`sync(func, lock?)`

把一个函数包装成“同一时刻只允许一个线程进去执行”的同步函数。

参数

参数	类型	可填值	默认值	说明
`func`	`function`	任意函数	无	要包装的函数，必填
`lock`	`lock`	`threads.lock()` 返回的锁对象	自动新建锁	可选共享锁

返回值

function

说明

如果没传 lock，会自动给你创建一把新的 ReentrantLock。
如果多个同步函数想共享同一把锁，你就手动把同一个 lock 传进去。
包装后的函数会先 lock.lock()，执行完后再自动 unlock()。

示例

const lock = threads.lock();
let count = 0;

const addOne = sync(function () {
  count += 1;
  return count;
}, lock);

threads.start(function () {
  log(addOne());
});

threads.start(function () {
  log(addOne());
});

`disposable.blockedGet(timeoutMs?)`

阻塞等待别人调用 setAndNotify(...) 填值。

参数

参数	类型	可填值	默认值	说明
`timeoutMs`	`number`	任意数字	`0`	`0` 表示一直等

返回值

any

说明

如果在超时前收到了值，就返回那个值。
如果超时了还没收到值，会返回当前值；初始情况下通常就是 null。
等待过程中会响应脚本暂停和脚本中断。

示例

const box = threads.disposable();

threads.start(function () {
  sleep(300);
  box.setAndNotify("done");
});

log(box.blockedGet(1000)); // done

`disposable.setAndNotify(value)`

设置值并唤醒所有正在等待的人。

参数

参数	类型	说明
`value`	`any`	要投递出去的值

返回值

undefined

说明

值会先做一次运行时解包，再存进容器。
这个容器没有“队列”概念，它只保存一份当前值。

示例

const box = threads.disposable();
box.setAndNotify({ ok: true });
log(JSON.stringify(box.blockedGet()));

`atomic.get()`

读当前原子值。

const atomic = threads.atomic(3);
log(atomic.get()); // 3

`atomic.set(value)`

直接把原子值改成 value。

atomic.set(10);

`atomic.getAndSet(value)`

先返回旧值，再改成新值。

log(atomic.getAndSet(20)); // 旧值

`atomic.incrementAndGet()`

先加 1，再返回新值。

log(atomic.incrementAndGet());

`atomic.decrementAndGet()`

先减 1，再返回新值。

log(atomic.decrementAndGet());

`atomic.getAndIncrement()`

先返回旧值，再加 1。

log(atomic.getAndIncrement());

`atomic.getAndDecrement()`

先返回旧值，再减 1。

log(atomic.getAndDecrement());

`atomic.addAndGet(delta)`

把当前值加上 delta，返回新值。

log(atomic.addAndGet(5));

`atomic.getAndAdd(delta)`

先返回旧值，再加上 delta。

log(atomic.getAndAdd(5));

`atomic.compareAndSet(expect, update)`

只有当前值等于 expect 时才更新成 update。

返回值

boolean

示例

if (atomic.compareAndSet(10, 99)) {
  log("updated");
}

`atomic.lazySet(value)`

延后可见地设置一个值。

说明

它也是 Java AtomicLong 的原生方法。大多数脚本场景下，如果你不熟并发内存语义，优先用普通 set(...) 就够了。

atomic.lazySet(7);

`atomic.toString()`

把当前原子值转成字符串。

log(atomic.toString());

`lock.lock()`

阻塞直到拿到锁。

lock.lock();
try {
  log("entered");
} finally {
  lock.unlock();
}

`lock.unlock()`

释放锁。

说明

通常应当放在 finally 里，避免中途抛错导致锁一直不放。

`lock.tryLock()`

尝试立刻拿锁，不等。

返回值

boolean

if (lock.tryLock()) {
  try {
    log("got lock");
  } finally {
    lock.unlock();
  }
}

`lock.lockInterruptibly()`

等待锁，但允许在线程被中断时提前退出等待。

说明

这是 Java 原生锁方法。只有你确实需要“可中断等待锁”时再用；普通脚本更常用 lock()。

`lock.newCondition()`

基于当前这把锁创建一个条件变量对象。

返回值

condition

示例

const lock = threads.lock();
const condition = lock.newCondition();

`lock.isLocked()`

判断这把锁当前是否被任何线程持有。

返回值

boolean

log(lock.isLocked());

`lock.isHeldByCurrentThread()`

判断当前线程是否持有这把锁。

返回值

boolean

log(lock.isHeldByCurrentThread());

`condition.await()`

释放当前锁并进入等待，直到被唤醒后再重新拿回锁。

说明

调用前必须已经持有对应的 lock。
这是 Java 条件变量原生语义。

示例

lock.lock();
try {
  condition.await();
} finally {
  lock.unlock();
}

`condition.awaitNanos(nanosTimeout)`

按纳秒超时等待条件。

参数

参数	类型	说明
`nanosTimeout`	`number`	最多等待多少纳秒

返回值

number

Java 语义里它返回剩余纳秒数。

`condition.awaitUntil(deadline)`

等到某个绝对时间点，或者等到被唤醒。

参数

参数	类型	说明
`deadline`	Java `Date`	截止时间

返回值

boolean

示例

imports("java.util.Date");

lock.lock();
try {
  const ok = condition.awaitUntil(new Date(System.currentTimeMillis() + 1000));
  log(ok);
} finally {
  lock.unlock();
}

`condition.signal()`

唤醒 1 个正在这个条件上等待的线程。

示例

lock.lock();
try {
  condition.signal();
} finally {
  lock.unlock();
}

`condition.signalAll()`

唤醒所有正在这个条件上等待的线程。

示例

lock.lock();
try {
  condition.signalAll();
} finally {
  lock.unlock();
}

常见报错与原因

回调不是函数

threads.start(action) requires a function
Thread.setTimeout(callback, delay[, ...args]) requires a function callback
Thread.setInterval(callback, delay[, ...args]) requires a function callback
Thread.setImmediate(callback[, ...args]) requires a function callback
setImmediate requires a function callback
sync(func) requires a function

这说明你把必填回调漏掉了，或者传进去的不是函数。

线程还没启动完就挂任务

thread has not started yet

这说明你在线程还没完成启动时就调用了 thread.setTimeout(...)、thread.setInterval(...) 或 thread.setImmediate(...)。
最稳的写法是先 thread.waitFor()。

线程已经结束

thread has already finished

这说明你往一个已经执行完或已经被中断的线程上继续挂任务了。

脚本正在退出

script exiting