java - 防止对对象的并发修改

Question

我在 Apache Camel 中有 2 个路由，MultimapRoute(根据数据库查询的结果在程序启动时初始化 Multimap 对象)和 UpdatingRoute(用于更新数据库的监听器，其中更新 Multimap 对象)当数据库更新时)。我想确保 MultimapRoute 首先运行，并且 UpdatingRoute 中的所有更改都按顺序处理。我还想确保所有并发问题都得到解决。

这是 MultimapRoute 类:

public class MultimapRoute implements org.apache.camel.Processor {
    static Multimap<String, String> m = new ArrayListMultimap<String, String>();
    static ReentrantLock l = new ReentrantLock();

    public void process(Exchange exchange) throws Exception {
        try {
            l.lock();
            //Query the database and fill out the map initially
        } finally {
            if (l.isHeldByCurrentThread()) l.unlock();
        }
    }
}

这是 UpdatingRoute 类:

public class UpdatingRoute implements org.apache.camel.Processor {
    public void process(Exchange exchange) throws Exception {
        try {
            l.lock();
            //Update the database based on latest event
        } finally {
            if (MultimapRoute.l.isHeldByCurrentThread())
                MultimapRoute.l.unlock();
        }
    }
}

我不确定这种多线程方法的线程安全或其他问题是什么，或者如何使程序线程安全。但我被告知这不是处理这个特定应用程序多线程的正确方法。非常感谢您的帮助。

Answer 1

首先，您需要在两个路由之间以某种方式进行通信来实现此目的:

I would like to ensure that the MultimapRoute runs first and all the changes from the UpdatingRoute are processed in order.

为此，我建议使用基于 ReentrantLock 的 Condition。

现在，问题的第二部分:

I'm not sure what the thread safety, or other, problem is with this approach to multithreading, or how to make the program thread-safe.

首先，您需要清楚地了解您希望使其成为线程安全的到底是什么意思。我的猜测 - 当某些线程成功启动 process() 方法(我的意思是，获取锁)时，您希望确保没有线程能够访问映射 m 。为此，请尝试更改访问 map 的方式。我的意思是，现在没问题，但是对 map 的引用在这两个类之外是可用的。这是我的建议，并实现了这两种想法:

public class MultimapRoute implements org.apache.camel.Processor {
    private static Multimap<String, String> m = new ArrayListMultimap<String, String>();
    private static ReentrantLock l = new ReentrantLock();
    public static boolean routeFinishedFlag = false;
    public static final Condition multimapRouteFinished = l.newCondition();

    public void process(Exchange exchange) throws Exception {
        l.lock();
        try {
            //Query the database and load some data to newData object
            performMapUpdate(newData);
            routeFinishedFlag = true;
            multimapRouteFinished.signal(); 
        } catch (InterruptedException e) {
            //if we're here, it means someone interrupted the thread that invoked process() method while it was trying to acquire map lock
        } finally {
            l.unlock();
        }
    }

    public static void performMapUpdate(Object newData) throws InterruptedException {
        l.lock();
        try {
            //here do some magic with the map and newData object. You may need to change method signature, it is just an illustration of approach
        } finally {
            l.unlock();
        }
    }
}

public class UpdatingRoute implements org.apache.camel.Processor {
    public void process(Exchange exchange) throws Exception {
        l.lock();
        try {
            while (!routeFinishedFlag) {
                MultimapRoute.multimapRouteFinished.await();
            }
            //Update the database based on latest event - again, I assume you have newObject object
            MultimapRoute.performMapUpdate(newData);
        } catch (InterruptedException e) {
            //if we're here, it means someone interrupted the thread that invoked process() method while it was waiting for condition or trying to acquire map lock
        } finally {
             l.unlock();
        }
    }
}