diff --git a/.gitignore b/.gitignore index 1d9756f..8188490 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,7 @@ torrents/ # 个人脚本（不提交到仓库） scripts/ + +# Java / Gradle（examples-jni） +examples-jni/build/ +examples-jni/.gradle/ diff --git a/examples-jni/README.md b/examples-jni/README.md index bb2197e..215183f 100644 --- a/examples-jni/README.md +++ b/examples-jni/README.md @@ -6,16 +6,14 @@ ``` examples-jni/ -├── build.gradle # Gradle 构建脚本 +├── build.gradle ├── settings.gradle └── src/main/java/cn/lmcw/dht/ - ├── model/ - │ ├── TorrentInfo.java # 与 Rust TorrentInfo 一一对应 - │ ├── FileInfo.java # 与 Rust FileInfo 一一对应 - │ └── DHTOptions.java # 与 Rust DHTOptions 一一对应 - ├── DhtListener.java # 事件回调接口 - ├── DhtCrawlerJni.java # JNI 绑定类（native 方法声明） - └── DhtCrawlerExample.java # 可运行示例 + ├── model/ # DHTOptions, TorrentInfo, FileInfo + ├── DhtCrawler.java # 面向对象入口（推荐） + ├── DhtCrawlerJni.java # 包内 native 绑定 + ├── DhtListener.java + └── DhtCrawlerExample.java ``` ## 快速开始 @@ -33,8 +31,8 @@ examples-jni/ # Linux / macOS java -Djava.library.path=. -jar dht-crawler-jni-example-.jar -# Windows（将 dht_crawler.dll 放在同目录） -java -Djava.library.path=. -jar dht-crawler-jni-example-.jar +# Windows（PowerShell/CMD 需对 -D 参数加引号，否则会报找不到主类） +java "-Djava.library.path=." -jar dht-crawler-jni-example-.jar ``` ### 方式二：从源码编译并运行 @@ -44,7 +42,6 @@ java -Djava.library.path=. -jar dht-crawler-jni-example-.jar 在仓库**根目录**执行： ```bash -# 本机（Linux 产出 libdht_crawler.so，Windows 产出 dht_crawler.dll，macOS 产出 libdht_crawler.dylib） cargo build --release --features jni ``` @@ -55,10 +52,7 @@ cargo build --release --features jni 在本目录（`examples-jni/`）执行： ```bash -# 使用默认库路径（../target/release） gradle run - -# 自定义库路径 gradle run -Plib.path=/path/to/your/lib ``` @@ -66,28 +60,28 @@ gradle run -Plib.path=/path/to/your/lib ```bash gradle shadowJar -# 产物：build/libs/dht-crawler-jni-example-.jar ``` ## 在自己的项目中集成 -1. 将 `src/main/java/cn/lmcw/dht/` 下的文件复制到你的项目。 -2. 将对应平台的 so/dll/dylib 放入 `java.library.path` 可访问的目录。 -3. 确保 JVM 启动时加了 `-Djava.library.path=<路径>`。 +1. 复制 `cn/lmcw/dht/` 下源码（含 `model/`、`DhtCrawler`、`DhtCrawlerJni`、`DhtListener`）。 +2. 将对应平台的 so/dll/dylib 放入 `java.library.path`。 -## JNI 生命周期 +## API（面向对象） ```java -DHTOptions options = new DHTOptions().setPort(6881); -long handle = DhtCrawlerJni.createServer(options, listener); -DhtCrawlerJni.startServer(handle); -// ... 运行 ... -DhtCrawlerJni.stopServer(handle); -DhtCrawlerJni.destroyServer(handle); // 必须调用，释放 Rust 资源 +DhtCrawler crawler = DhtCrawler.createServer(options, listener); +crawler.start(); +// ... +crawler.stop(); // 或 try-with-resources ``` +- **`DhtCrawler.createServer(options, listener)`**：创建会话（未启动 DHT；Java 不能用方法名 `new`，故不用 `open`）。 +- **`start()`**：后台启动 DHT，非阻塞；同一会话多次 `start()` 仅首次生效。 +- **`stop()` / `close()`**：停止并释放 Rust 资源，幂等。 +- **`getNodePoolSize()`**：routing table 节点数。 + ## 注意事项 -- `destroyServer` 必须在不再使用后调用，否则 Rust 侧的 tokio runtime 和连接不会释放。 -- 回调方法（`onTorrent`、`onError`）在 Rust 的 tokio 工作线程中触发，请确保实现是线程安全的。 -- 同一个句柄不要在多个线程中并发调用 `destroyServer`。 +- 回调在 Rust 工作线程触发，实现需线程安全。 +- `onMetadataFetch` 在阻塞线程池调用，宜快速返回。 diff --git a/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawler.java b/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawler.java new file mode 100644 index 0000000..a562f04 --- /dev/null +++ b/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawler.java @@ -0,0 +1,100 @@ +package cn.lmcw.dht; + +import cn.lmcw.dht.model.DHTOptions; + +/** + * DHT 爬虫会话：封装 native 句柄与生命周期，推荐使用的面向对象入口。 + * + * 
{@code
+ * try (DhtCrawler crawler = DhtCrawler.createServer(options, listener)) {
+ *     crawler.start();
+ *     // ... 运行 ...
+ * } // close → stop，释放 Rust 资源
+ *
+ * // 或手动：
+ * DhtCrawler c = DhtCrawler.createServer(options, listener);
+ * c.start();
+ * // ...
+ * c.stop();
+ * }
+ */ +public final class DhtCrawler implements AutoCloseable { + + private long handle; + private volatile boolean started; + + private DhtCrawler(long handle) { + this.handle = handle; + } + + /** + * 创建 DHT 服务器会话（尚未 {@link #start()}，与 native createServer 对应）。 + * Java 中方法不能命名为 {@code new}，故用 {@code createServer}。 + * + * @param options 配置，{@code null} 使用 Rust 默认 + * @param listener 回调，{@code null} 不注册回调 + * @return 已绑定 native 资源的会话 + * @throws IllegalStateException 创建失败（句柄为 0，通常伴随 JVM 异常） + */ + public static DhtCrawler createServer(DHTOptions options, DhtListener listener) { + long h = DhtCrawlerJni.createServer(options, listener); + if (h == 0) { + throw new IllegalStateException("DHT 服务器创建失败（createServer 返回 0）"); + } + return new DhtCrawler(h); + } + + /** + * 在后台启动 DHT（非阻塞）。可多次调用，仅首次生效。 + * + * @throws IllegalStateException 已 {@link #stop()} 关闭后不能再启动 + */ + public synchronized void start() { + if (handle == 0) { + throw new IllegalStateException("会话已关闭，无法 start"); + } + if (started) { + return; + } + DhtCrawlerJni.startServer(handle); + started = true; + } + + /** + * 停止并释放全部 Rust 资源（tokio runtime 等）。幂等：重复调用安全。 + */ + public synchronized void stop() { + if (handle == 0) { + return; + } + DhtCrawlerJni.stopServer(handle); + handle = 0; + started = false; + } + + /** 是否已调用过 {@link #start()} 且尚未 {@link #stop()}。 */ + public synchronized boolean isStarted() { + return started && handle != 0; + } + + /** 会话仍有效（未 stop）时为 true。 */ + public synchronized boolean isOpen() { + return handle != 0; + } + + /** + * 当前 routing table 节点数；已关闭时返回 0。 + */ + public int getNodePoolSize() { + long h = handle; + if (h == 0) { + return 0; + } + return DhtCrawlerJni.getNodePoolSize(h); + } + + @Override + public void close() { + stop(); + } +} diff --git a/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java b/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java index 61f400b..c314fa4 100644 --- a/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java +++ b/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java @@ -24,10 +24,9 @@ import java.util.concurrent.atomic.AtomicLong; public class DhtCrawlerExample { public static void main(String[] args) throws InterruptedException { - // 1. 构造配置 DHTOptions options = new DHTOptions() .setPort(6881) - .setNetMode(0) // 0 = IPv4 Only + .setNetMode(0) .setMetadataTimeout(5L) .setMaxMetadataQueueSize(50_000) .setMaxMetadataWorkerCount(500); @@ -35,7 +34,6 @@ public class DhtCrawlerExample { AtomicLong torrentCount = new AtomicLong(); CountDownLatch shutdown = new CountDownLatch(1); - // 2. 实现回调 DhtListener listener = new DhtListener() { @Override public void onTorrent(TorrentInfo info) { @@ -49,34 +47,29 @@ public class DhtCrawlerExample { public void onError(String message) { System.err.println("[ERROR] " + message); } + + @Override + public boolean onMetadataFetch(String infoHash) { + return true; + } }; - // 3. 创建并启动服务器 - long handle = DhtCrawlerJni.createServer(options, listener); - if (handle == 0) { - System.err.println("创建服务器失败"); - return; - } - System.out.println("DHT 服务器已创建，正在启动..."); - DhtCrawlerJni.startServer(handle); - System.out.println("DHT 服务器已启动，端口 " + options.getPort()); + final DhtCrawler crawler = DhtCrawler.createServer(options, listener); + crawler.start(); + System.out.println("DHT 已启动，端口 " + options.getPort()); - // 4. 注册 JVM 关闭钩子，确保资源释放 Runtime.getRuntime().addShutdownHook(new Thread(() -> { System.out.println("\n收到退出信号，正在停止..."); - DhtCrawlerJni.stopServer(handle); - try { Thread.sleep(500); } catch (InterruptedException ignored) {} - DhtCrawlerJni.destroyServer(handle); - System.out.println("服务器已销毁，共发现种子：" + torrentCount.get()); + crawler.stop(); + System.out.println("已关闭，共发现种子：" + torrentCount.get()); shutdown.countDown(); })); - // 5. 定期打印节点池大小 System.out.println("按 Ctrl+C 退出。每 10 秒打印一次节点池大小..."); while (true) { Thread.sleep(10_000); - int poolSize = DhtCrawlerJni.getNodePoolSize(handle); - System.out.println("节点池大小: " + poolSize + " 已发现种子: " + torrentCount.get()); + System.out.println("节点池: " + crawler.getNodePoolSize() + + " 已发现种子: " + torrentCount.get()); } } } diff --git a/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java b/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java index 4800414..26e2249 100644 --- a/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java +++ b/examples-jni/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java @@ -3,21 +3,9 @@ package cn.lmcw.dht; import cn.lmcw.dht.model.DHTOptions; /** - * Rust DHT-Crawler 库的 JNI 绑定入口。 - * - *

所有方法均为 {@code static native}，通过 JNI 调用 Rust 实现。 - * 加载顺序由静态初始化块保证，使用前无需手动调用。

- * - *

生命周期

- * 
{@code
- * long handle = DhtCrawlerJni.createServer(options, listener);
- * DhtCrawlerJni.startServer(handle);
- * // ... 运行中 ...
- * DhtCrawlerJni.stopServer(handle);
- * DhtCrawlerJni.destroyServer(handle); // 必须调用，否则 Rust 资源泄漏
- * }
+ * 包内 JNI 绑定（由 {@link DhtCrawler} 使用）。业务代码请用 {@link DhtCrawler}。 */ -public final class DhtCrawlerJni { +final class DhtCrawlerJni { static { System.loadLibrary("dht_crawler"); @@ -25,44 +13,11 @@ public final class DhtCrawlerJni { private DhtCrawlerJni() {} - /** - * 创建并初始化 DHT 服务器，返回 Rust 侧句柄。 - * - * @param options 服务器配置，传 {@code null} 则使用 Rust 侧默认值 - * @param listener 事件回调，传 {@code null} 则不注册任何回调 - * @return 服务器句柄（非 0 表示成功），或 0（失败，同时会向 JVM 抛出异常） - */ - public static native long createServer(DHTOptions options, DhtListener listener); + static native long createServer(DHTOptions options, DhtListener listener); - /** - * 启动已创建的 DHT 服务器（非阻塞）。 - * 服务器在 Rust tokio runtime 后台运行，此方法立即返回。 - * - * @param handle {@link #createServer} 返回的句柄 - */ - public static native void startServer(long handle); + static native void startServer(long handle); - /** - * 向服务器发送停止信号（非阻塞）。 - * 调用后服务器将优雅退出，但资源尚未释放，需继续调用 {@link #destroyServer}。 - * - * @param handle {@link #createServer} 返回的句柄 - */ - public static native void stopServer(long handle); + static native void stopServer(long handle); - /** - * 销毁服务器并释放所有 Rust 侧资源（tokio runtime、连接等）。 - * 必须调用，否则发生内存泄漏。调用后不得再使用该句柄。 - * - * @param handle {@link #createServer} 返回的句柄 - */ - public static native void destroyServer(long handle); - - /** - * 获取当前节点池（routing table）中的节点数量。 - * - * @param handle {@link #createServer} 返回的句柄 - * @return 节点数量，句柄无效时返回 0 - */ - public static native int getNodePoolSize(long handle); + static native int getNodePoolSize(long handle); } diff --git a/examples-jni/src/main/java/cn/lmcw/dht/DhtListener.java b/examples-jni/src/main/java/cn/lmcw/dht/DhtListener.java index fb8a335..701bc78 100644 --- a/examples-jni/src/main/java/cn/lmcw/dht/DhtListener.java +++ b/examples-jni/src/main/java/cn/lmcw/dht/DhtListener.java @@ -22,4 +22,17 @@ public interface DhtListener { * @param message Rust 侧错误的文本描述 */ void onError(String message); + + /** + * 在即将对某 info_hash 拉取 BitTorrent metadata 之前调用。 + * 与 Rust {@code DHTServer::on_metadata_fetch} 对应：返回 {@code true} 才会真正去拉取； + * 返回 {@code false} 则跳过该 hash，不会触发 {@link #onTorrent}。 + *

在 Rust 的 metadata 工作线程中通过阻塞线程池调用，应快速返回，避免长时间阻塞。

+ * + * @param infoHash 40 字符小写十六进制 info_hash + * @return 是否允许拉取 metadata + */ + default boolean onMetadataFetch(String infoHash) { + return true; + } } diff --git a/jni/callbacks.rs b/jni/callbacks.rs index 23f00fa..ff02f39 100644 --- a/jni/callbacks.rs +++ b/jni/callbacks.rs @@ -4,12 +4,13 @@ use crate::jni_bindings::types::torrent_info_to_java; use jni::objects::JValue; use std::sync::Arc; -/// 向 `DHTServer` 注册所有 Java 回调（on_torrent、on_error）。 +/// 向 `DHTServer` 注册所有 Java 回调（on_torrent、on_error、on_metadata_fetch）。 /// /// `callback` 持有 Java listener 的全局引用，可跨线程安全使用。 pub fn register_callbacks(server: &Arc, callback: JavaCallback) { register_on_torrent(server, callback.clone()); - register_on_error(server, callback); + register_on_error(server, callback.clone()); + register_on_metadata_fetch(server, callback); } /// 注册 on_torrent 回调：Rust TorrentInfo → Java listener.onTorrent(TorrentInfo) @@ -54,3 +55,37 @@ fn register_on_error(server: &Arc, callback: JavaCallback) { } }); } + +/// 注册 on_metadata_fetch：在拉取 metadata 前询问 Java `onMetadataFetch(infoHash)`。 +/// 返回 `true` 才继续拉取；在阻塞线程池中执行 JNI，避免阻塞 tokio worker。 +fn register_on_metadata_fetch(server: &Arc, callback: JavaCallback) { + server.on_metadata_fetch(move |info_hash| { + let cb = callback.clone(); + async move { + let r = tokio::task::spawn_blocking(move || { + cb.with_env(|env, listener| { + let j_s = env.new_string(&info_hash)?; + let out = env.call_method( + listener, + "onMetadataFetch", + "(Ljava/lang/String;)Z", + &[JValue::Object(&j_s.into())], + )?; + Ok(out.z()?) + }) + }) + .await; + match r { + Ok(Ok(allow)) => allow, + Ok(Err(e)) => { + log::error!("回调 onMetadataFetch 失败: {e}，默认允许拉取"); + true + } + Err(e) => { + log::error!("onMetadataFetch spawn_blocking 失败: {e}，默认允许拉取"); + true + } + } + } + }); +} diff --git a/jni/exports.rs b/jni/exports.rs index 3916a12..4dcfb56 100644 --- a/jni/exports.rs +++ b/jni/exports.rs @@ -83,7 +83,8 @@ pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_startServer( }); } -/// 停止 DHTServer（发送关闭信号，不释放资源）。 +/// 停止并销毁 DHTServer：发关闭信号后释放 tokio runtime 等全部 Rust 资源。 +/// 调用后句柄失效，勿再传入任何 native 方法。 /// /// Java 签名：`native void stopServer(long handle);` #[unsafe(no_mangle)] @@ -92,39 +93,22 @@ pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_stopServer( _class: JClass, handle: jlong, ) { - crate::jni_catch!(&mut env, (), { - let h = unsafe { + jni_shutdown_and_release(&mut env, handle); +} + +fn jni_shutdown_and_release(env: &mut JNIEnv, handle: jlong) { + crate::jni_catch!(env, (), { + if handle == 0 { + return; + } + unsafe { match handle_ref(handle) { - Some(h) => h, + Some(h) => h.stop(), None => { let _ = env.throw_new("java/lang/IllegalArgumentException", "无效的服务器句柄"); return; } } - }; - h.stop(); - }); -} - -/// 销毁 DHTServer：停止并释放所有 Rust 资源（包括 tokio runtime）。 -/// 调用后 Java 侧不得再使用该句柄。 -/// -/// Java 签名：`native void destroyServer(long handle);` -#[unsafe(no_mangle)] -pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_destroyServer( - mut env: JNIEnv, - _class: JClass, - handle: jlong, -) { - crate::jni_catch!(&mut env, (), { - if handle == 0 { - return; - } - // 先停止，再释放 - unsafe { - if let Some(h) = handle_ref(handle) { - h.stop(); - } destroy_handle(handle); } });