Java Review - 使用Event Bus实现目录变化的监控
Pre
Java Review - Java进程内部的消息中间件_Event Bus设计模式
需求
监控文件的实时变化 , 就这一句话？？？
分析一下其中一个有问题的思路：
程序首次启动时获取该文件的最后修改时间并且做文件的首次解析，然后每隔一段指定的时间检查一次文件最后被修改的时间，如果与记录的时间相等则等待下次的采集（Balking Pattern），否则进行新一轮的采集并且更新时间。
这。。。。有问题啊 ， 比如在采集时间间隔内，文件发生了N次变化，只能获取到最后一次，其根本原因是文件的变化不会通知到应用程序， 我只能傻傻的轮询~
新思路：
JDK自1.7版本后提供了WatchService类，该类可以基于事件通知的方式监控文件或者目录的任何变化，文件的改变相当于每一个事件（Event）的发生，针对不同的时间执行不同的动作，结合NIO2.0中提供的WatchService和Event Bus实现文件目录的监控的功能。
Event Bus案例实战
import com.artisan.bfzm.eventbus.EventBus;
import java.nio.file.*;
/**
* @author 小工匠
* @version 1.0
* @description: TODO
* @date 2021/12/2 16:37
* @mark: show me the code , change the world
*/
public class DirectoryTargetMonitor {
private WatchService watchService;
private final EventBus eventBus;
private final Path path;
private volatile boolean start = false;
public DirectoryTargetMonitor(final EventBus eventBus,
final String targetPath) {
this(eventBus, targetPath, "");
}
/**
*
* 构造Monitor的时候需要传入EventBus以及需要监控的目录
* @param eventBus
* @param targetPath
* @param morePaths
*/
public DirectoryTargetMonitor(final EventBus eventBus,
final String targetPath, final String... morePaths) {
this.eventBus = eventBus;
this.path = Paths.get(targetPath, morePaths);
}
public void startMonitor() throws Exception {
this.watchService = FileSystems.getDefault().newWatchService();
//为路径注册感兴趣的事件
this.path.register(watchService,
StandardWatchEventKinds.ENTRY_MODIFY,
StandardWatchEventKinds.ENTRY_DELETE,
StandardWatchEventKinds.ENTRY_CREATE);
System.out.printf("The directory [%s] is monitoring... \n", path);
this.start = true;
while (start) {
WatchKey watchKey = null;
try {
//当有事件发生时会返回对应的WatchKey
watchKey = watchService.take();
watchKey.pollEvents().forEach(event ->
{
WatchEvent.Kind<?> kind = event.kind();
Path path = (Path) event.context();
Path child = DirectoryTargetMonitor.this.path.resolve(path);
//提交FileChangeEvent到EventBus
eventBus.post(new FileChangeEvent(child, kind));
});
} catch (Exception e) {
this.start = false;
} finally {
if (watchKey != null) {
watchKey.reset();
}
}
}
}
public void stopMonitor() throws Exception {
System.out.printf("The directory [%s] monitor will be stop...\n", path);
Thread.currentThread().interrupt();
this.start = false;
this.watchService.close();
System.out.printf("The directory [%s] monitor will be stop done.\n", path);
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
在创建WatchService之后将文件的修改、删除、创建等注册给了WatchService，在指定目录下发生诸如此类的事件之后便会收到通知，我们将事件类型和发生变化的文件Path封装成FileChangeEvent提交给Event Bus.
import java.nio.file.Path;
import java.nio.file.WatchEvent;
/**
* @author 小工匠
* @version 1.0
* @description: TODO
* @date 2021/12/2 16:38
* @mark: show me the code , change the world
*/
public class FileChangeEvent {
private final Path path;
private final WatchEvent.Kind<?> kind;
public FileChangeEvent(Path path, WatchEvent.Kind<?> kind) {
this.path = path;
this.kind = kind;
}
public Path getPath() {
return path;
}
public WatchEvent.Kind<?> getKind() {
return kind;
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
FileChangeEvent是对WatchEvent.Kind和Path的包装，一旦目录发生任何改变，都会提交FileChangeEvent事件。
目录监控的程序我们已经实现了，下面就来写一个接受文件目录变化的Subscriber，也就是当目录发生变化时用来接受事件的方法
import com.artisan.bfzm.eventbus.Subscribe;
/**
* @author 小工匠
* @version 1.0
* @description: TODO
* @date 2021/12/2 16:39
* @mark: show me the code , change the world
*/
public class FileChangeListener {
@Subscribe
public void onChange(FileChangeEvent event) {
System.out.printf("%s-%s-%s\n", Thread.currentThread().getName(),event.getPath(), event.getKind());
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
onChange方法由@Subscribe标记，但没有指定topic，当有事件发送到了默认的topic上之后，该方法将被调用执行，接下来我们将FileChangeListener的实例注册给Event Bus并且启动Monitor程序
import com.artisan.bfzm.eventbus.AsyncEventBus;
import com.artisan.bfzm.eventbus.EventBus;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
/**
* @author 小工匠
* @version 1.0
* @description: TODO
* @date 2021/12/2 16:39
* @mark: show me the code , change the world
*/
public class FileDirTest {
public static void main(String[] args) throws Exception {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
final EventBus eventBus = new AsyncEventBus(executor);
//注册
eventBus.register(new FileChangeListener());
DirectoryTargetMonitor monitor = new DirectoryTargetMonitor(eventBus, "C:\\Users\\artisan\\Desktop\\aaa");
monitor.startMonitor();
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
小结
EventBus有点类似于GOF设计模式中的监听者模式，但是EventBus提供的功能更加强大，使用起来也更加灵活，EventBus中的Subscriber不需要继承任何类或者实现任何接口，在使用EventBus时只需要持有Bus的引用即可。
在EventBus的设计中有三个非常重要的角色（Bus、Registry和Dispatcher），Bus主要提供给外部使用的操作方法，Registry注册表用来整理记录所有注册在EventBus上的Subscriber，Dispatcher主要负责对Subscriber消息进行推送（用反射的方式执行方法），但是考虑到程序的灵活性，Dispatcher方法中又提供了Executor的多态方式。
文章来源: artisan.blog.csdn.net，作者：小小工匠，版权归原作者所有，如需转载，请联系作者。
原文链接：artisan.blog.csdn.net/article/details/122356501
- 点赞
- 收藏
- 关注作者
评论（0）