如何使用 EJB 3.0 实现适当的计数器 bean？

Question

[编辑] 这个问题是“我如何使用 EJB 3 和 JPA 2.0 对实体 bean 进行原子更改”。应该很简单吧？

我尝试根据目前得到的答案修复我的代码。我正在使用 JBoss 6.0.0M2 和 Hypersonic（只需下载它并调用 run.bat）。

我的测试用例：创建 3 个线程并在循环中调用其中一个 testCounterMitLock*() 500 次。所以一个成功的测试应该打印出“Anzahl eingetragene Zeilen: 1500” (3*500)。

我试过了：

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
        manager.lock(ct, LockModeType.WRITE);
        int wert = ct.getWert();

显然不起作用，因为不同的线程可以在应用锁之前更改数据库中的值。所以我尝试解决这个问题：

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
        manager.lock(ct, LockModeType.WRITE);
        manager.refresh (ct);
        int wert = ct.getWert();

refresh() 应该给我当前值，隐式查询也应该确保对象现在被锁定。没有这样的运气。让我们试试 JPA 2.0：

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.WRITE);
        int wert = ct.getWert();

这也行不通。也许锁不够？

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
        int wert = ct.getWert();

嗯……也不行！最后一次绝望的尝试：

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
        manager.flush();
        manager.refresh (ct);
        int wert = ct.getWert();

好的...谁能解释为什么没有任何效果？我没有想法。

[EDIT2] PS：雪上加霜，这是最后一个运行线程的最后输出：

commit/rollback: 441/62

(441+62 = 503)...

这是完整的代码。首先是bean：

package server.kap15;

import java.rmi.RemoteException;

import javax.ejb.*;
import javax.persistence.*;

@Stateful
public class CounterTestBean implements CounterTestRemote, SessionSynchronization {
    @PersistenceContext(unitName = "JavaEE")
    EntityManager manager;

    private int commit = 0;

    private int rollback = 0;

    public void initDatenbank() {
        manager.createNamedQuery("CounterTest.deleteAll").executeUpdate();
        manager.createNamedQuery("TestTabelle.deleteAll").executeUpdate();
        CounterTestVersion ct = new CounterTestVersion();
        ct.setNr(1);
        ct.setVersion(1);
        ct.setWert(1);
        manager.persist(ct);
    }

    public boolean testCounterOhneLock() {
        try {
            CounterTest ct = manager.find(CounterTest.class, 1);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
            manager.lock(ct, LockModeType.WRITE);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock2() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1);
            manager.lock(ct, LockModeType.WRITE);
            manager.refresh (ct);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock3() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.WRITE);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock4() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitLock5() {
        try {
            CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
            manager.flush();
            manager.refresh (ct);
            int wert = ct.getWert();
            ct.setWert(wert + 1);
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (Throwable t) {
            return false;
        }
    }

    public boolean testCounterMitVersion() {
        try {
            CounterTestVersion ctv = manager.find(CounterTestVersion.class, 1);
            int wert = ctv.getWert();
            ctv.setWert(wert + 1);
            manager.flush();
            TestTabelle tt = new TestTabelle();
            tt.setNr(wert);
            manager.persist(tt);
            manager.flush();
            return true;
        } catch (OptimisticLockException e) {
            System.out.println(">>> Versionskonflikt !");
            return false;
        } catch (Throwable t) {
            System.out.println(t.getMessage());
            return false;
        }
    }

    public long anzTestZeilen() {
        Query query = manager.createNamedQuery("TestTabelle.anzZeilen");
        Long anzahl = (Long) query.getSingleResult();
        return anzahl;
    }

    public void afterBegin() throws EJBException, RemoteException {
    }

    public void beforeCompletion() throws EJBException, RemoteException {
    }

    public void afterCompletion(boolean committed) throws EJBException,
    RemoteException {
        if (committed)
            commit++;
        else
            rollback++;
        System.out.println("commit/rollback: " + commit + "/" + rollback);
    }
}

远程接口：

package server.kap15;

import javax.ejb.Remote;

@Remote
public interface CounterTestRemote {
    public void initDatenbank();

    public boolean testCounterOhneLock();

    public boolean testCounterMitLock();
    public boolean testCounterMitLock2();
    public boolean testCounterMitLock3();
    public boolean testCounterMitLock4();
    public boolean testCounterMitLock5();

    public boolean testCounterMitVersion();

    public long anzTestZeilen();
}

persistence.xml：

<?xml version="1.0" encoding="UTF-8"?>
<persistence xmlns="http://java.sun.com/xml/ns/persistence"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/persistence
        http://java.sun.com/xml/ns/persistence/persistence_1_0.xsd"
    version="1.0">
    <persistence-unit name="JavaEE">
        <jta-data-source>java:DefaultDS</jta-data-source>
    </persistence-unit>
</persistence>

测试客户端：

package client.kap15;

import java.util.Properties;
import javax.naming.*;
import javax.rmi.PortableRemoteObject;
import server.kap15.CounterTestRemote;

public class CounterTestMitLock extends Thread {
    CounterTestRemote ctr;

    public static void main(String[] args) {
        try
        {
            testMitLock();
            testMitLock2();
            testMitLock3();
            testMitLock4();
            testMitLock5();
        }
        catch (Exception e)
        {
            e.printStackTrace ();
        }
    }

    static int N = 3;
    static CounterThread[] ct = new CounterThread[N];

    private static void testMitLock () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock();

        runTest ();
    }

    private static void testMitLock2 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock2 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock2();

        runTest ();
    }

    private static void testMitLock3 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock3 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock3();

        runTest ();
    }

    private static void testMitLock4 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock4 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock4();

        runTest ();
    }

    private static void testMitLock5 () throws InterruptedException
    {
        System.out.println("--- Counter Test MIT Lock5 ----------------------");
        System.out.println("Testinstanzen erzeugen...");
        for (int i=0; i<N; i++)
            ct[i] = new CounterThreadMitLock5();

        runTest ();
    }

    private static void runTest () throws InterruptedException
    {
        System.out.println("Datenbank initialisieren...");
        ct[0].ctr.initDatenbank();

        System.out.println("Test durchführen...");
        for (int i=0; i<N; i++)
            ct[i].start();

        System.out.println("Auf Ende warten...");
        for (int i=0; i<N; i++)
            ct[i].join();

        System.out.println("Anzahl eingetragene Zeilen: " + ct[0].ctr.anzTestZeilen());
    }

    private static CounterTestRemote verbinden() {
        try {
            Properties p = new Properties();
            p.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
            p.put(Context.URL_PKG_PREFIXES, "org.jboss.naming:org.jnp.interfaces");
            p.put(Context.PROVIDER_URL, "jnp://localhost:1099");
            Context ctx = new InitialContext(p);

            Object ref = ctx.lookup("CounterTestBean/remote");
            CounterTestRemote ctr = (CounterTestRemote) PortableRemoteObject.narrow(ref, CounterTestRemote.class);

            return ctr;
        } catch (NamingException e) {
            System.out.println("ERROR - NamingException!");
            System.exit(-1);
        }
        return null;
    }

    public abstract static class CounterThread extends Thread
    {
        protected CounterTestRemote ctr;

        public CounterThread ()
        {
            this.ctr = verbinden ();
        }

        public void run() {
            for (int i = 0; i < 500; i++)
                test ();
        }

        public abstract void test ();
    }

    public static class CounterThreadMitLock extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock();
        }

    }

    public static class CounterThreadMitLock2 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock2();
        }

    }

    public static class CounterThreadMitLock3 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock3();
        }

    }

    public static class CounterThreadMitLock4 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock4();
        }

    }

    public static class CounterThreadMitLock5 extends CounterThread
    {
        @Override
        public void test ()
        {
            this.ctr.testCounterMitLock5();
        }

    }
}

Answer 1

由于没有一种锁定模式起作用，我尝试了ewernli 的手动解决方案SELECT ... FOR UPDATE。这给出了一个有趣的例外：“意外的令牌 FOR”。于是我查看了数据库。

JBoss 默认安装Hypersonic 1.8 (HSQLDB)，不支持行锁定。亲爱的 JBoss 开发人员：当不支持锁定模式时，JPA 实现应该抛出异常。

所以我添加了一个 Oracle 数据源并更改了我的 persistence.xml。之后，两个测试工作：

        CounterTestVersion ct = manager.find(CounterTestVersion.class, 1, LockModeType.PESSIMISTIC_WRITE);
        int wert = ct.getWert();

和

    Query query = manager.createNativeQuery ("select * from COUNTER_TEST where NR = 1 for update", CounterTestVersion.class);
    CounterTestVersion ct = (CounterTestVersion)query.getSingleResult ();
    int wert = ct.getWert ()+1;

这很有趣。它也应该适用于LockModeType.PESSIMISTIC_FORCE_INCREMENT。在这种情况下，我在日志中看到了这个错误：

ORA-00054: resource busy and acquire with NOWAIT specified

这发生在调用manager.find() 中。我不明白为什么两者在加载阶段表现不同。可能是 JBoss 或 Hibernate 中的错误。

Answer 2

我有几点意见：

• 您当前正在使用乐观锁定，但我在您的实体上看不到任何@Version 字段。我认为这行不通。
• 如果您希望计数器递增 1500 次，请不要使用乐观锁定（您不希望任何更新被 OptimisticLockingException 拒绝）而是悲观锁定。
• 吞下Throwable 确实是错误的，你希望容器完成他的工作（但我想你知道这一点）。

所以，我会在这里改用这个：

manager.lock(ct, LockModeType.READ);

并删除catch (Throwable t)。

更新：我现在无法测试，但我会使用类似的东西（其余代码不变）：

public boolean testCounterWithLock() {
    CounterTest ct = manager.find(CounterTest.class, 1);
    manager.lock(ct, LockModeType.READ);
    int counter = ct.getCounter();
    ct.setCounter(counter + 1);
    manager.flush();
    return true;
}

我真的怀疑这会奏效。首先，读锁不会阻止其他线程更新行。其次，另一个线程可以更新 find() 和 getCounter() 之间的行

你是对的，我走得太快了，上面肯定不是一个解决方案，@ewernli 也是对的，JPA 1.0 不支持悲观锁定策略，你必须依赖数据库（并使用SELECT FOR UPDATE 语义）。不知何故，我设法忘记了这一点，并对 READ 模式产生了很大的困惑。我的错。感谢您指出这一点。

我认为您必须使用 LockModeType.WRITE 但也许您可以在 lock() 之后使用 em.refresh() 以确保实体不会过时？

使用LockModeType.WRITE时，在UPDATE的WHERE子句中添加@Version注解的实体字段，并在UPDATE期间进行并发检查：

UPDATE COUNTERTEST SET COUNTER = ?, OPT_LOCK = ? 
WHERE ((ID = ?) AND (OPT_LOCK = ?))

如果WHERE 子句匹配记录失败（因为另一个线程已经更新了实体），那么持久性提供程序将抛出OptimisticLockException。

换句话说，在lock() 之后刷新实体不会改变任何东西，另一个线程仍然可以刷新同一个实体，而另一个线程正在修改计数器。以自动化方式处理乐观锁定的唯一方法是实现重试机制。

但是当flush() 抛出PersitenceException（NoResultException 和NonUniqueResultException 的实例除外）时，当前事务被标记为回滚，因此不能用于事务目的。所以每次重试都必须使用一个新的事务来执行。在无状态 bean 中，您可以进行递归远程调用，但我认为这在有状态 bean 中没有意义，因此您必须从客户端处理。

最后，这并不是很令人满意，在我看来，在 JPA 1.0 中处理这个问题的不太糟糕的方法是使用 SELECT FOR UPDATE 获得锁。

Answer 3

即使使用LockModeType.READ 或LockModeType.WRITE，JPA 1.0 也仅支持乐观锁定。锁定获取仍然可以推迟到提交时间，因此您遇到了问题。

来自JPA 2.0 concurrency and locking：

仅支持 PA 1.0 乐观读取 或乐观写锁定。 JPA 2.0 支持乐观和悲观 锁定

其他资源：EJB3 performance 和 Pessimist Locking with JPA

要使用 JPA 1.0 实现真正的悲观锁定，您需要依赖数据库或特定于实现的扩展。例如：

JPA 2.0（Hibernate API 可以实现类似的功能）

Account acc = em.find( Account.class, id, PESSIMISTIC );

JPA 1.0

Query query = em.createNativeQuery("SELECT * ... FOR UPDATE"); // works with most db
Account acc = (Account) query.getSingleResult();

至少，这是我最终使用的，因为lock 没有按预期工作。

( 注意：当乐观异常发生时，您也可以实现重试逻辑。但这很复杂，因为事务由应用程序服务器管理。您需要使用@TRANSACTION_NEW 来暂停当前事务并开始新的事务一个，等等……我觉得太复杂了！）

Answer 4

你没有展示你对 testCounterWithLock 的返回值做了什么。我的猜测是您遇到了乐观锁定失败，并且有时返回值为 false。

当冲突在实践中可能很少发生时，优化锁定是一个合理的模型，调用者可以合理地重做工作。所以如果你得到乐观的失败，你可以重试。

或者，使用悲观锁定模型，这会在您读取的点锁定数据库中的行。你可以在调用 find() 时添加一个悲观的 LockMode。悲观锁的使用需要小心谨慎，很容易导致并发性差和/或死锁。