2

我正在将 Spring JMS 与 JPA Hibernate 实现结合使用，我看到插入时出现间歇性问题，然后即时读取同一记录。

网络应用流程：

- 数据被发布到我的 Web 应用程序 Web 服务，数据被发送到 Glassfish OpenMQ 队列（下面的 STUInputQ）。
-com.api.listener.backoffice.STUMessageListener 读取 STUInputQ 队列并插入到我们的 Oracle 数据库中，然后将消息（带有新的数据库主键）发送到另一个队列（下面的 ArchiveQ）。
-com.api.listener.backoffice.StorableMessageListener 读取 ArchiveQ 队列并尝试使用 com.api.listener.backoffice.STUMessageListener 插入的数据库记录的主键读取数据库。

问题：

有时（大约 18%）StorableMessageListener 中的读取操作会返回 null，即使该记录确实存在。在我看来，即使插入返回序列生成的主键，在读取发生之前插入提交也没有处理。我在插入数据的方法和读取数据的方法的末尾放置了一个 unix 时间戳，当问题发生时，unix 时间戳是相同的，所以看起来好像读取在最终提交之前获得了消息.

临时解决方案：

我添加了一些逻辑来使线程休眠，并确保在读取数据库时我永远不会得到空值。我真的不认为线程睡眠是一个长期的解决方案。关于为什么 STUMessageListener 在 StorableMessageListener 读取之前无法提交事务的任何想法？

依赖项：

休眠核心.3.3.2.GA

休眠实体管理器-3.4.0.GA

春天 3.0.6.RELEASE

Java 1.5

弹簧配置：

<?xml version="1.0" encoding="UTF-8"?>
            <beans xmlns="http://www.springframework.org/schema/beans"
                xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                xmlns:util="http://www.springframework.org/schema/util"
                xmlns:context="http://www.springframework.org/schema/context"
                xmlns:aop="http://www.springframework.org/schema/aop"
                xmlns:tx="http://www.springframework.org/schema/tx"
                xmlns:hz="http://www.hazelcast.com/schema/spring"
                xsi:schemaLocation="
                http://www.hazelcast.com/schema/spring
                http://www.hazelcast.com/schema/spring/hazelcast-spring-2.5.xsd
                http://www.springframework.org/schema/beans
                http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
                http://www.springframework.org/schema/util
                http://www.springframework.org/schema/util/spring-util-3.0.xsd
                http://www.springframework.org/schema/context
                http://www.springframework.org/schema/context/spring-context-3.0.xsd
                http://www.springframework.org/schema/tx
                http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
                http://www.springframework.org/schema/aop
                http://www.springframework.org/schema/aop/spring-aop-3.0.xsd">

                <!-- Generic -->
                <context:annotation-config />
                <context:component-scan base-package="myapp.api" />
                <aop:aspectj-autoproxy/>

                <!-- JPA -->
                <bean class="org.springframework.orm.jpa.support.PersistenceAnnotationBeanPostProcessor" />

                <tx:annotation-driven />

                <bean class="org.springframework.dao.annotation.PersistenceExceptionTranslationPostProcessor" />

                <bean id="entityManagerFactory" class="org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean">
                    <property name="dataSource" ref="dataSource" />
                    <property name="jpaVendorAdapter" ref="jpaVendorAdapter" />
                    <property name="persistenceUnitName" value="MyApp" />
                    <property name="jpaProperties">
                        <props>
                            <prop key="hibernate.use_sql_comments">true</prop>
                            <prop key="hibernate.generate_statistics">true</prop>
                            <prop key="hibernate.archive.autodetection">class</prop>
                            <prop key="hibernate.cache.use_second_level_cache">true</prop>
                            <prop key="hibernate.cache.provider_class">com.hazelcast.hibernate.provider.HazelcastCacheProvider</prop>
                            <prop key="hibernate.cache.use_query_cache">true</prop>
                            <prop key="hibernate.cache.use_minimal_puts">true</prop>
                        </props>
                    </property>
                </bean>


               <hz:hazelcast id="instance">
                    <hz:config>
                         //rest of Hazelast config maps here
                    </hz:config>
                </hz:hazelcast>

                <hz:hibernate-region-factory id="regionFactory" instance-ref="instance"/>

                <!-- Define JPA Provider Adapter -->
                <bean id="jpaVendorAdapter" class="org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter">
                    <property name="showSql" value="true" />
                    <property name="generateDdl" value="true" />
                    <property name="databasePlatform" value="org.hibernate.dialect.OracleDialect" />
                </bean>

                <bean id="dataSourceTarget" class="oracle.jdbc.pool.OracleDataSource" destroy-method="close">
                    <property name="URL" value="jdbc:oracle:thin:@server:1525:name" />
                    <property name="user" value="test" />
                    <property name="password" value="123" />
                    <property name="connectionCachingEnabled" value="true" />
                    <property name="connectionCacheProperties">
                        <props merge="default">
                            <prop key="MinLimit">5</prop>
                            <prop key="MaxLimit">50</prop>
                        </props>
                    </property>
                </bean>

                <bean id="dataSource" class="org.springframework.jdbc.datasource.LazyConnectionDataSourceProxy">
                    <property name="targetDataSource" ref="dataSourceTarget"/>
                </bean>


                <bean id="transactionManager" class="org.springframework.orm.jpa.JpaTransactionManager">
                    <property name="entityManagerFactory" ref="entityManagerFactory"/>
                    <property name="dataSource" ref="dataSource"/>
                </bean>

                <tx:annotation-driven transaction-manager="transactionManager" proxy-target-class="false"/>

                <bean id="genericDAO" class="myapp.api.dao.impl.GenericDAOImpl">
                    <constructor-arg>
                        <value>java.io.Serializable</value>
                    </constructor-arg>
                </bean>

                <bean id="springContextHolder" class="myapp.api.util.SpringContextHolder" factory-method="getInstance" />

        <bean id="executionInterceptor" class="myapp.api.listener.backoffice.ExecutionInterceptor" />

        <!-- JNDI-->
        <bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate"/>

        <!-- JMS -->
        <bean id="jmsQueueConnectionFactory"  class="org.springframework.jndi.JndiObjectFactoryBean">
            <property name="jndiTemplate">
                <ref bean="jndiTemplate"/>
            </property>
            <property name="jndiName" value="${jms.jndi.qconnectionfactory}">

            </property>
        </bean>
        <bean id="myJMSConnectionFactory" class="com.api.model.vo.backoffice.OpenMqConnectionFactoryBean">
            <property name="imqAddressList" value="${jms.imq.url}" />
            <property name="imqDefaultUsername" value="${jms.imq.user}" />
            <property name="imqDefaultPassword" value="${jms.imq.password}" />
            <property name="imqHost" value="${jms.imq.host}" />
            <property name="imqPort" value="${jms.imq.port}" />
        </bean>

        <bean id="stuMessageListener" class="com.api.listener.backoffice.STUMessageListener" />
        <bean id="storeListener" class="com.api.listener.backoffice.StorableMessageListener"/>

        <bean id="executionInterceptor" class="com.api.listener.backoffice.ExecutionInterceptor" />

        <bean id="stuJmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="jmsQueueConnectionFactory"/>
            <property name="destinationName" value="STUInputQ"/>
            <property name="sessionTransacted" value="false"/>
            <property name="messageListener" ref="stuMessageListener" />
            <property name="concurrentConsumers" value="5" />
            <property name="maxConcurrentConsumers" value="100" />
            <property name="receiveTimeout" value="30000" />
            <property name="cacheLevelName" value="CACHE_NONE" />

        </bean>
        <bean id="storeJmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="jmsQueueConnectionFactory"/>
            <property name="destinationName" value="ArchiveQ"/>
            <property name="sessionTransacted" value="false"/>
            <property name="messageListener" ref="storeListener" />
            <property name="concurrentConsumers" value="5" />
            <property name="maxConcurrentConsumers" value="100" />
            <property name="receiveTimeout" value="30000" />
            <property name="cacheLevelName" value="CACHE_NONE" />

        </bean>

 </beans>

持久化配置：

<?xml version="1.0" encoding="UTF-8"?>
<persistence version="1.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_1_0.xsd">
  <persistence-unit name="com" transaction-type="RESOURCE_LOCAL">
  </persistence-unit>
</persistence>

插入记录的类：

public class STUMessageListener implements javax.jms.MessageListener{
       @Autowired
       StoringService storingService;

       @Transactional
       public void onMessage(Message message) throws RuntimeException { 
           try {
                Object omsg = ((ObjectMessage) message).getObject();
                if (omsg instanceof StorableMessage) {
                     StorableMessage storableMessage = (StorableMessage) omsg;
                     //StorableMessage insert into Database
                     storingService.store(storableMessage); 

                   //jms logic here to send message to next queue (ArchiveQ)
           }
           catch (Throwable ex) {
                throw new RuntimeException(ex);
            }
   }

     @Service("storingService")
        public class StoringServiceImpl{

            @Autowired
            MessagesDAO messagesDAO;

            @Transactional
            public StorableMessage store(StorableMessage storableMessage) {
                messagesDAO.save(storableMessage);
            }

        }

    @Repository("messagesDAO")
    public class MessagesDAOImpl{
        private Class<T> type
        @PersistenceContext
        EntityManager entityManager;

        public void save(T object) {
           entityManager.persist(object);
        }

        public T findById(Serializable id) {
           return entityManager.find(type, id);
        }
    }

读取数据库记录的类：

public class StorableMessageListener implements javax.jms.MessageListener {
    @Autowired
    MessageDAO messageDAO;

    @Transactional
    public void onMessage(Message message) throws RuntimeException {
          if (omsg instanceof StorableMessage) {
               //this is where null is returned for the Messages object 18% of the time
               //sleep thread by 1 second logic here helps eliminate the null Messages object
              //uses same MessageDAO as above 
              Messages msg = messageDAO.findById(storableMessage.getMessageKey());
          }
    }
4

1 回答 1

0

尝试将插入方法的注释更改为

@Transactional（传播=传播.REQUIRES_NEW）

这将在方法完成后立即提交插入。

于 2013-05-24T10:47:25.957 回答