JDK 之 NIO 2 WatchServiceWatchKey（监控文件变化）

Posted 2021-01-31 binarylei

JDK 之 NIO 2 WatchService、WatchKey（监控文件变化）

JDK 规范目录(https://www.cnblogs.com/binarylei/p/10200503.html)

一、WatchService、WatchKey 使用

具体详见：https://blog.csdn.net/lirx_tech/article/details/51425364

public static void main(String[] args) throws Exception {
    // 1. 获取文件系统监控器，启动一个后台线程轮询
    WatchService watchService = FileSystems.getDefault().newWatchService();

    // 2. 注册要监听的事件类型，文件增、删、改
    Paths.get("C:\\Users\\len\\Desktop\\xdr").register(watchService,
            StandardWatchEventKinds.ENTRY_CREATE,
            StandardWatchEventKinds.ENTRY_DELETE,
            StandardWatchEventKinds.ENTRY_MODIFY);

    while (true) {
        // 3. 获取准备好的事件，pool() 立即返回、take() 阻塞
        WatchKey watchKey = watchService.poll(2, TimeUnit.SECONDS);
        if (Objects.isNull(watchKey)) {
            continue;
        }
        // 4. 处理准备好的事件
        List<WatchEvent<?>> watchEvents = watchKey.pollEvents();
        for (WatchEvent<?> event : watchEvents) {
            if (event.kind().name().equals(StandardWatchEventKinds.ENTRY_CREATE.name())) {
                System.out.println("create: " + event.context());
            } else if (event.kind().name().equals(StandardWatchEventKinds.ENTRY_MODIFY.name())) {
                System.out.println("modify: " + event.context());
            } else if (event.kind().name().equals(StandardWatchEventKinds.ENTRY_DELETE.name())) {
                System.out.println("delete: " + event.context());
            }
        }
        // 5. 重启该线程，因为处理文件可能是一个耗时的过程，因此调用 pool() 时需要阻塞监控器线程
        boolean valid = watchKey.reset();
        if (!valid) {
            break;
        }
    }
}

二、原理

• AbstractWatchService 实现 WatchService 接口
• WindowsWatchService 具体的实现，启动 Poller 线程
• Poller 线程，轮询指定的目录

(1) AbstractWatchService

// 等待要处理的事件 signaled keys waiting to be dequeued
private final LinkedBlockingDeque<WatchKey> pendingKeys = new LinkedBlockingDeque<WatchKey>();

@Override
public final WatchKey poll(long timeout, TimeUnit unit) throws InterruptedException {
    checkOpen();
    WatchKey key = pendingKeys.poll(timeout, unit);
    checkKey(key);
    return key;
}

可以看到调用 poll 的时候直接从队列中取 key，那就必然有一个线程往 pendingKeys 中塞数据。在 AbstractWatchService 中有一个 enqueueKey 方法往 pendingKeys 中塞数据。

final void enqueueKey(WatchKey key) {
    pendingKeys.offer(key);
}

(2) WindowsWatchService

AbstractWatchService 有不同的实现，以 WindowsWatchService 为例。

WindowsWatchService(WindowsFileSystem fs) throws IOException {
    // create I/O completion port
    long port = 0L;
    try {
        port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0);
    } catch (WindowsException x) {
        throw new IOException(x.getMessage());
    }
    this.poller = new Poller(fs, this, port);
    this.poller.start();
}

Poller 是 WindowsWatchService 的内部类，开启了一个线程监控目录。

(3) Poller

重点关注 Poller 中的 run 方法。

@Override
public void run() {
    for (;;) {
        CompletionStatus info;
        try {
            info = GetQueuedCompletionStatus(port);
        } catch (WindowsException x) {
            return;
        }

        WindowsWatchKey key = ck2key.get((int)info.completionKey());
        if (key == null) {
            continue;
        }
        boolean criticalError = false;
        if (errorCode == ERROR_NOTIFY_ENUM_DIR) {
            key.signalEvent(StandardWatchEventKinds.OVERFLOW, null);
        } else if (errorCode != 0 && errorCode != ERROR_MORE_DATA) {
            criticalError = true;
        } else {
            // 省略... (处理 error)
        }

        // 一切正常则 criticalError = true，此时将这个 WatchKey 加入 pendingKeys 中 
        if (criticalError) {
            implCancelKey(key);
            key.signal();
        }
    }
}

参考：

1. 《NIO.2：WatchService、WatchKey（监控文件变化）》：(https://blog.csdn.net/lirx_tech/article/details/51425364)

---

每天用心记录一点点。内容也许不重要，但习惯很重要！