JTA 事务和线程

Question

我想用 JTA 做一个多线程处理。

环境：

• JBOSS AS 6.3
• Java 7
• Oracle 11.2g
• CDI

治疗：我想用数据库中的所有数据制作一个 zip。这个 zip 文件可能很大，所以我想在 jboss 将它发送到客户端的同时启动一个线程来生成流。

我的 REST 条目：

@Stateless
@Path("/exportProcess")
public class ExportProcessusResource {

    @Inject
    private IExport export;

    @GET
    @Path("/{processCode: [^/]+}")
    @Produces(MediaType.APPLICATION_OCTET_STREAM)
    public Response export(@PathParam("processCode") final String pProcessCode) {
        return Response.ok(export.export(pProcessCode))
                                .header("Content-Disposition", "attachment; filename=" + pCodeProcessus + ".zip")
                                .build();
    }
}

我的模型：

@Entity
@Table(name = "T_PROCESS")
@NamedQueries({
    @NamedQuery(name = "Process.GetByCode", query = "SELECT p FROM Process p WHERE p.code=:code")
})
public class Process {

    @Column(name = "CODE", length = 50, nullable = false)
    private String code;

    @OneToMany(mappedBy = "process", targetEntity = Step.class)
    private Collection<Step> steps;

    //Getters/Setters
}

@Entity
@Table(name = "T_STEP")
public class STEP {
    @Id
    @Column(name = "ID_STEP")
    @GeneratedValue(strategy = GenerationType.AUTO, generator = "SEQ_ID_STEP")
    @SequenceGenerator(name = "SEQ_ID_STEP", sequenceName = "SEQ_ID_STEP")
    private int id;

    @ManyToOne(targetEntity = Process.class)
    @JoinColumn(name = "CODE_PROCESS", referencedColumnName = "CODE", nullable = false)
    private Process process;

    //Getters/Setters
}

我的 DAO：

public interface IProcessDao {

    Processus getByCode(final String pCode);
}

public class ProcessDao implements IProcessDao {

    @Override
    public Processus getByCode(final String pCode) {
        Processus lResult = null;
        try {
            final TypedQuery<Processus> lRequest = pEm.createNamedQuery("Process.GetByCode", Process.class);
            lRequest.setParameter("code", pCode);
            lResult = lRequest.getSingleResult();
        } catch (final NoResultException e) {
            // Return null
            lResult = null;
        }
        return lResult;
    }
}

我的控制器：

public interface IExport {

    /**
     * Generate export
     * 
     * @param pProcessCode Process code
     * @return Datas
     */
    InputStream export(final String pProcessCode);
}


public class Export implements IExport {

    @PersistenceContext(unitName="authorizations")
    private EntityManager entityManagerAuthorizations;

    @Inject
    private ExportThreadHelper exportThreadHelper;

    @Override
    public InputStream export(final String pProcessCode) {
        //Check if user has the profile. Use database "AUTHORIZATIONS"
        checkProfil(entityManagerAuthorizations, Profiles.ADMIN);

        final PipedInputStream lInputStream = new PipedInputStream();
        OutputStream lOutputStream = null;
        try {
            lOutputStream = new FileOutputStream("d:/test.zip");// new
                                                                // PipedOutputStream(lInputStream);
        } catch (final IOException e) {
            throw new RuntimeException("Cannot start zip generation", e);
        }

        final ZipOutputStream lZipOutputStream = new ZipOutputStream(lOutputStream);

        final Runnable lRunnable = new Runnable() {
            @Override
            public void run() {
                try {
                    exportThreadHelper.export(pProcessCode, lZipOutputStream);
                } catch (final Exception e) {
                    logger.error(e);
                } finally {
                    IOUtils.closeQuietly(lZipOutputStream);
                }
            }
        };
        //To execute in same thread :
        //lRunnable.run();

        //To execute in another thread
        final Thread lThread = new Thread(lRunnable);
        lThread.start();
        try {
            lThread.join();
        } catch (final InterruptedException e1) {
            throw new RuntimeException(e1);
        }


        try {
            return new FileInputStream("d:/test.zip");
        } catch (final FileNotFoundException e) {
            logger.error(e);
        }
        return lInputStream;
    }
}


public class ExportThreadHelper {

    private class ProcessToExport {
        //...
    }

    @PersistenceContext
    @Named("Application")
    private EntityManager entityManagerThreadable;

    @Inject
    private IProcessDao processDao;

    public void export(final String pProcesssCode, final ZipOutputStream pZipOutputStream)
                    throws MyWayBusinessException {
        try {

            final ProcessToExport lProcessToExport = new ProcessToExport();

            transaction(entityManagerThreadable, new Callable<Void>() {
                @Override
                public Void execute() {
                    final Process lProcess = processDao.getByCode(pProcesssCode);
                    for (final Step lStep : lProcess.getSteps()) {
                        //Many things
                    }
                    return null;
                }
            });

            //MANY OTHER TREATMENTS

        } catch (final Exception e) {
            logger.error(e);
            throw new RuntimeException("Cannot generate export", e);
        }
    }

    @Override
    @TransactionAttribute(TransactionAttributeType.REQUIRED)
    protected <T> T transaction(final EntityManager pEntityManager, final Callable<T> pCallable) {
        //I've tried with and without the annotation and with and without the "UserTransaction"
        try {
            final UserTransaction tx = com.arjuna.ats.jta.UserTransaction.userTransaction();
            try {
                tx.begin();
                final T lResultat = pCallable.execute();
                tx.commit();
                return lResultat;
            } catch (final Throwable e) {
                tx.rollback();
                throw e;
            }
        } catch (final Throwable e) {
            throw new RuntimeException(e);
        }
    }
}

我的persistence.xml：

<?xml version="1.0" encoding="UTF-8"?>
<persistence version="2.0"
    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_2_0.xsd ">
    <persistence-unit name="APPLICATION" transaction-type="JTA">
        <jta-data-source>java:/jdbc/app</jta-data-source>
        <class>Processus</class>
        <class>Step</class>
        <properties>
            <!-- Scan for annotated classes and Hibernate mapping XML files -->
            <property name="hibernate.archive.autodetection" value="class, hbm" />
        </properties>
    </persistence-unit>
    <persistence-unit name="AUTHORIZATION" transaction-type="JTA">
        <jta-data-source>java:/jdbc/AUTHORIZATION</jta-data-source>
        <!-- many things... -->
    </persistence-unit>
</persistence>

（我已经清理了代码，只保留重要的东西）。

而且，如果我使用单线程版本 (lRunnable.run())，我有 zip 文件，但如果我运行多线程版本 (thread.start())（我已在此处阻止以确保我的测试连接没有被父线程关闭，但是在我删除 thread.join()) 之后我有这个异常：

错误 [...ExportThreadHelper] (Thread-115) 延迟初始化失败 角色集合：.steps，无法初始化代理 - 没有会话： org.hibernate.LazyInitializationException：延迟初始化失败 角色集合：.steps，无法初始化代理 - 没有会话 在 org.hibernate.collection.internal.AbstractPersistentCollection.throwLazyInitializationException（AbstractPersistentCollection.java:569） [hibernate-core-4.2.14.SP1-redhat-1.jar:4.2.14.SP1-redhat-1] 在 org.hibernate.collection.internal.AbstractPersistentCollection.withTemporarySessionIfNeeded(AbstractPersistentCollection.java:188) [hibernate-core-4.2.14.SP1-redhat-1.jar:4.2.14.SP1-redhat-1] 在 org.hibernate.collection.internal.AbstractPersistentCollection.initialize(AbstractPersistentCollection.java:548) [hibernate-core-4.2.14.SP1-redhat-1.jar:4.2.14.SP1-redhat-1] 在 org.hibernate.collection.internal.AbstractPersistentCollection.read(AbstractPersistentCollection.java:126) [hibernate-core-4.2.14.SP1-redhat-1.jar:4.2.14.SP1-redhat-1] 在 org.hibernate.collection.internal.PersistentBag.iterator(PersistentBag.java:266) [hibernate-core-4.2.14.SP1-redhat-1.jar:4.2.14.SP1-redhat-1] 在 ExportThreadHelper$1.execute(ExportThreadHelper.java:101) [metier-2.3.0-SNAPSHOT.jar:] 在 ExportThreadHelper$1.execute(ExportThreadHelper.java:1) [metier-2.3.0-SNAPSHOT.jar:] 在 ExportThreadHelper.transaction(ExportThreadHelper.java:148) [metier-2.3.0-SNAPSHOT.jar:] 在 ExportThreadHelper.export(ExportThreadHelper.java:97) [metier-2.3.0-SNAPSHOT.jar:] 在 ExportMetier$1.run(ExportMetier.java:62) [metier-2.3.0-SNAPSHOT.jar:] 在 java.lang.Thread.run(Thread.java:722) [rt.jar:1.7.0_04]

您在我的代码中发现问题了吗？

Answer 1

您需要像这样更改查询：

@NamedQueries({
    @NamedQuery(name = "Process.GetByCode", query = "SELECT p FROM Process p LEFT JOIN FETCH p.steps WHERE p.code=:code")
})