java 文件锁 文件序列化 多进程并发处理 工具类

该类提供文件并发操作的控制，适用于多进程并发时通过文件锁来同步操作。
提供：

• 对文件上锁并支持传入一个consumer进行操作，在文件上锁期间，执行传入的consumer，执行完毕后，自动关闭文件锁。
• 支持传入一个锁竞争失败后，执行失败回调处理。
• 对象序列化，并支持多进程并发控制序列化到文件操作。

package com.baiyang.walnut.utils;import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.io.*;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;/*** 该类提供文件并发操作的控制，适用于多进程并发时通过文件锁来同步操作。* 提供：* - 对文件上锁并支持传入一个consumer进行操作，在文件上锁期间，执行传入的consumer，执行完毕后，自动关闭文件锁。* - 支持传入一个锁竞争失败后，执行失败回调处理。* - 对象序列化，并支持多进程并发控制序列化到文件操作。* * 详见实现。* * User: Yang Peng*/
public class FileUtils {/*** 提供对文件的写锁上锁与释放，调用方只需要关注consumer的实现即可。* LockConsumer会传入LockInfo，lockInfo封装了path对应的输出流，consumer直接使用即可。* consumer无需手动关闭流，在外部会自动关闭。** @author Yang Peng*/public static <R> R writeLockShell(@NotNull String path, @NotNull LockConsumer<R, WriteLockInfo> handler) throws Exception {return writeLockShell(path, false, handler, null);}public static <R> R writeLockNonBlockShell(@NotNull String path, @NotNull LockConsumer<R, WriteLockInfo> handler, @NotNull LockConflictConsumer<WriteLockInfo> lockConflictConsumer) throws Exception {return writeLockShell(path, false, handler, lockConflictConsumer);}public static <R> R writeLockShell(@NotNull String path, boolean append, @NotNull LockConsumer<R, WriteLockInfo> handler) throws Exception {return writeLockShell(path, append, handler, null);}/*** 提供对文件的写锁上锁与释放，调用方只需要关注consumer的实现即可。* LockConsumer会传入LockInfo，lockInfo封装了path对应的输出流，consumer直接使用即可。* consumer无需手动关闭流，在外部会自动关闭。** @author Yang Peng*/public static <R> R writeLockShell(@NotNull String path, boolean append, @NotNull LockConsumer<R, WriteLockInfo> handler,LockConflictConsumer<WriteLockInfo> lockConflictConsumer) throws Exception {// 注意！FileOutputStream需要采用append模式，否则在上锁之前会清空文件，导致并发读取文件读取不到内容。// 如果要控制并发操作，则需要在对文件内容进程操作之前，先获取锁才是正确操作。// 所以采用append模式来达到上述目的，这只是变通方式（未找到合适的方法），在对文件进行操作时，还需要truncate(0)来清空文件try (FileOutputStream fileOutputStream = new FileOutputStream(path, true)) {try (FileChannel channel = fileOutputStream.getChannel()) {WriteLockInfo writeLockInfo = new WriteLockInfo(fileOutputStream, path, channel);if (lockConflictConsumer != null) { //如果传入了锁冲突处理，则表示采用非阻塞模式FileLock lock = channel.tryLock(); //当前如果其他进程占用着锁，不阻塞，获取的锁为nullif (lock == null) { //如果未获取到锁lockConflictConsumer.accept(writeLockInfo); //锁冲突处理return null;}if (!append)channel.truncate(0);//清空文件return handler.accept(writeLockInfo);} else {//阻塞模式，直到获取到锁try (FileLock lock = channel.lock()) {if (!append)channel.truncate(0);//清空文件return handler.accept(writeLockInfo);}}}}}/*** 提供对文件的读锁上锁与释放，调用方只需要关注consumer的实现即可。* LockConsumer会传入LockInfo，lockInfo封装了path对应的输入流，consumer直接使用即可。* consumer无需手动关闭流，在外部会自动关闭。** @author Yang Peng*/public static <R> R readLockShell(@NotBlank String path, @NotNull LockConsumer<R, ReadLockInfo> consumer) throws Exception {try (FileInputStream fileInputStream = new FileInputStream(path)) {try (FileChannel channel = fileInputStream.getChannel()) {try (FileLock lock = channel.lock(0L, Long.MAX_VALUE, true)) {ReadLockInfo readLockInfo = new ReadLockInfo(fileInputStream, path, channel);return consumer.accept(readLockInfo);}}}}/*** 将对象序列化到文件。* 写入动作可通过flock参数上锁。上锁采用tryLock，如果未获取到锁，则返回false，并不会执行写入动作，由上层来决定如何处理。** @param path  序列化文件路径，如果目录不存在会报错* @param obj   对象，需实现Serializable接口* @param flock 是否上文件锁，用于防止多进程对文件并发操作导致文件数据破坏。* @return 是否上锁成功，也可以表示是否写入成功。true-写入成功，false-上锁失败(未写入文件)*/public static boolean writeObj(@NotBlank String path, @NotNull Object obj, boolean flock) throws Exception {// FileOutputStream需要采用append模式，否则在上锁之前会清空文件，导致并发读取文件读取不到内容。// 如果要控制并发操作，则需要在对文件内容进程操作之前，先获取锁才是正确操作。// 所以采用append模式来达到上述目的，这只是变通方式（未找到合适的方法），在对文件进行操作时，还需要truncate(0)来清空文件CeOptional<Boolean> optional = new CeOptional<>(true);if (flock) {//非阻塞模式writeLockNonBlockShell(path, item -> {ObjectOutputStream outputStream = new ObjectOutputStream(item.fileOutputStream);outputStream.writeObject(obj);return null;}, item -> {optional.set(false); //锁冲突处理});} else {try (FileOutputStream fileOutputStream = new FileOutputStream(path);ObjectOutputStream outputStream = new ObjectOutputStream(fileOutputStream)) {outputStream.writeObject(obj);}}return optional.get();}/*** 读取对象序列化文件，并加载到对象中。* 读取动作可通过flock参数上读锁。** @param path  序列化文件路径，如果目录不存在会报错* @param c     对象类型。* @param flock 是否上文件锁，用于防止多进程对文件并发操作导致文件数据破坏。*/public static <T> T readObj(@NotBlank String path, @NotNull Class<T> c, boolean flock) throws Exception {T obj = null;if (flock) {obj = readLockShell(path, item -> {ObjectInputStream inputStream = new ObjectInputStream(item.getFileInputStream());return (T) inputStream.readObject();});} else {try (FileInputStream fileInputStream = new FileInputStream(path);ObjectInputStream inputStream = new ObjectInputStream(fileInputStream)) {obj = (T) inputStream.readObject();} catch (EOFException e) {//文件末尾}}return obj;}public static interface LockConsumer<R, L extends LockInfo> {R accept(L t) throws Exception;}public static interface LockConflictConsumer<L extends LockInfo> {void accept(L t) throws Exception;}public static class LockInfo {private final FileChannel fileChannel;private final String path;public LockInfo(String path, FileChannel fileChannel) {this.path = path;this.fileChannel = fileChannel;}public FileChannel getFileChannel() {return fileChannel;}public String getPath() {return path;}}public static class WriteLockInfo extends LockInfo {private final FileOutputStream fileOutputStream;public WriteLockInfo(FileOutputStream fileOutputStream, String path, FileChannel fileChannel) {super(path, fileChannel);this.fileOutputStream = fileOutputStream;}public FileOutputStream getFileOutputStream() {return fileOutputStream;}}public static class ReadLockInfo extends LockInfo {private final FileInputStream fileInputStream;public ReadLockInfo(FileInputStream fileInputStream, String path, FileChannel fileChannel) {super(path, fileChannel);this.fileInputStream = fileInputStream;}public FileInputStream getFileInputStream() {return fileInputStream;}}/*** can edit optional* User: Yang Peng*/public static class CeOptional<T> {private T value;public CeOptional() {}public CeOptional(T value) {this.value = value;}public T get() {return value;}public void set(T value) {this.value = value;}}
}

用法示例：

String path = "/xx/xx/xx.xx";
//给某个操作处理，包裹上文件锁，多进程并发控制，处理完毕自动释放文件锁
String writeFilePath = writeLockNonBlockShell(path, (fileInfo) -> {//文件流FileOutputStream fileOutputStream = fileInfo.getFileOutputStream();//通道FileChannel fileChannel = fileInfo.getFileChannel();//文件路径String filePath = fileInfo.getPath();System.out.println("文件上锁成功，并处理..");//返回任意对象return filePath;
}, (fileInfo) -> {//文件流FileOutputStream fileOutputStream = fileInfo.getFileOutputStream();FileChannel fileChannel = fileInfo.getFileChannel();String filePath = fileInfo.getPath();System.out.println("当前文件被其他进程锁定，锁获取失败处理...");
});//给某个操作处理，包裹上文件锁，多进程并发控制，处理完毕自动释放文件锁
String readFilePath = readLockShell(path, fileInfo -> {//文件流FileInputStream fileInputStream = fileInfo.getFileInputStream();//通道FileChannel fileChannel = fileInfo.getFileChannel();//文件路径String filePath = fileInfo.getPath();System.out.println("文件上锁成功，并处理..");return filePath;
});//将对象序列化到文件
writeObj(path, new Object(), true);
//序列化文件读取对象
Object obj = readObj(path, Object.class, true);

使用请充分测试。