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

- 日理万妓 2022-10-22 15:56 9阅读 0赞

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

*  对文件上锁并支持传入一个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(); //当前如果其他进程占用着锁，不阻塞，获取的锁为null
                        if (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);

使用请充分测试。