多线程：基于java中资源的动态对象锁定

Question

这里需要线程专家的眼睛......

我在一个 poc 应用程序上，我在 FTP 服务器上上传文件

在 FTP 服务器中有多个文件夹。根据输入响应，我从文件夹中读取文件并移动到另一个文件夹

应用可以同时被多个线程访问。

所以问题是这样的：

假设 FTP 有一个文件夹 Folder_A 和 A_A_FOLDER 现在 Folder_A 有 10 个文件。 一个线程来了，从 FTP 读取了 10 个文件并开始对其进行一些计算， 它一一计算，然后移动到 A_A_FOLDER 它处于过程的中间（假设它成功地将 5 个文件从 Folder_A 移动到 A_A_FOLDER） 然后另一个线程来了，它选择了剩余的 5 个文件，因为它们被线程 1 处理不足，所以线程 2 也开始处理这 5 个文件

所以这里有重复文件的问题

void m1(String folderName) {
// FTP related code
}

我已经通过使用同步关键字解决了这个问题

现在一切都在同步，所有处理工作正常

synchronized void m1(String folderName) {
// code
}

文件夹名称决定需要处理哪个文件夹

现在我开始面临性能问题

因为方法是同步的，所以所有线程都会等到处理线程未完成其任务。

我可以通过以下步骤改进：

（在找到解决方案之前，这里有一些关于这个问题的故事）

正如我提到的 m1 方法的 folderName 参数决定将处理哪个文件夹， 所以假设我在 Ftp 服务器中有 4 个文件夹（A、B、A_T、B_T），2 个文件夹是需要从（A 和 B）读取数据的文件夹， 2个文件夹是数据将移动的文件夹（A_T和B_T）

A_T 和 B_T 不是这里的问题，因为它们对于每个文件夹 A 和 B 都是唯一的 因此，如果该方法将从 A 读取，那么它会将其移至 A_T，与 B 相同（移至 B_T）

现在：

假设 m1 方法有 4 个线程，文件夹 A 有 3 个线程，文件夹 B 有 1 个线程 如果基于 fileName 参数以某种方式同步请求以便我可以提高性能，则意味着 1 个线程将在 A 上工作，另外 2 个线程将阻塞，因为它们的 fileName 相同，所以他们将等到第一个线程未完成它的任务，线程 4 将并行无需任何锁定过程即可工作，因为它的文件名不同

那么我怎样才能在代码级别实现这一点（在文件名上同步）？

注意：我知道我可以使用资源的静态锁定列表然后锁定文件名资源来打破这个逻辑 例如：

private final Object A = new Object();
private final Object B = new Object();

但这种方法的问题是文件夹可以动态添加，所以我不能这样做。

需要你们的帮助。

Answer 1

一种方法是为每个目录维护一个锁：

public class DirectoryTaskManager {
    public static void main(String[] args) throws IOException {
        DirectoryTaskManager manager = new DirectoryTaskManager();
        manager.withDirLock(new File("Folder_A"), () -> System.out.println("Doing something..."));
    }

    public void withDirLock(File dir, Runnable task) throws IOException {
        ReentrantLock lock = getDirLock(dir);
        lock.lock();
        try {
            task.run();
        } finally {
            lock.unlock();
        }
    }

    private Map<File, ReentrantLock> dirLocks = Collections.synchronizedMap(new HashMap<>());

    public ReentrantLock getDirLock(File dir) throws IOException {
        // Resolve the canonical file here so that different paths 
        // to the same file use the same lock
        File canonicalDir = dir.getCanonicalFile();
        if (!canonicalDir.exists() || !canonicalDir.isDirectory()) {
            throw new FileNotFoundException(canonicalDir.getName());
        }
        return dirLocks.computeIfAbsent(canonicalDir, d -> new ReentrantLock());
    }
}

Answer 2

感谢@teppic 和@OlegSklyar 的指导 最后这里是完整的工作示例，

FolderImpl -> 有方法名调用，可以被多个线程访问

我使用了 ConcurrentHashMap（读取可以非常快，而写入是通过锁定完成的。）它比 synchronizedMap 快 它将保存文件夹名称和 ReentrantLock，因此锁定将对文件夹名称起作用

public class FolderImpl {
    private FolderImpl(){
        System.out.println("init................");
    }

    private ConcurrentHashMap<String, ReentrantLock> concurrentHashMap= new ConcurrentHashMap();
    private static final FolderImpl singleTon = new FolderImpl();
    public static FolderImpl getSingleTon() {
        return singleTon;
    }

    public void call(String name) throws Exception{
        ReentrantLock getDirLock = getDirLock(name);
        getDirLock.lock();
        try {
        for (int i = 0; i < 100; i ++) {
            System.out.println(i+":"+name+":"+Thread.currentThread().getName());
            try {
                Thread.sleep(30);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }}finally {
            getDirLock.unlock();
        }

    }

    public ReentrantLock getDirLock(String site)  {
        return concurrentHashMap.computeIfAbsent(site, d -> new ReentrantLock());
    }
}

TaskCaller 线程调用 call 方法，这里是 sleep 风格，所以另一个线程可以 git 执行时间

public class TaskCaller extends Thread{
    public FolderImpl singleTon = FolderImpl.getSingleTon();
    public TaskCaller(String name) {
        super();
        this.name = name;
    }

    private String name;
    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            System.out.println("Name:"+Thread.currentThread().getName());

            try {
                singleTon.call(name);
                sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

TestExecution 类将执行 10 个线程进行测试

public class TestExecution {

    public static void main(String[] args) {
        TaskCaller testThreadCC = new TaskCaller("A_FOLDER");
        TaskCaller testThreadCC2 = new TaskCaller("A_FOLDER");
        TaskCaller testThreadCC3 = new TaskCaller("B_FOLDER");
        TaskCaller testThreadCC4 = new TaskCaller("C_FOLDER");
        TaskCaller testThreadCC5 = new TaskCaller("C_FOLDER");
        TaskCaller testThreadCC6 = new TaskCaller("C_FOLDER");
        TaskCaller testThreadCC7 = new TaskCaller("A_FOLDER");
        TaskCaller testThreadCC8 = new TaskCaller("A_FOLDER");
        TaskCaller testThreadCC9 = new TaskCaller("B_FOLDER");
        TaskCaller testThreadCC10 = new TaskCaller("B_FOLDER");

        testThreadCC.start();
        testThreadCC2.start();
        testThreadCC3.start();
        testThreadCC4.start();
        testThreadCC5.start();
        testThreadCC6.start();
        testThreadCC7.start();
        testThreadCC8.start();
        testThreadCC9.start();
        testThreadCC10.start();

    }

}