Spring singleton bean - 后台线程；没有具有实际交易可用错误的 EntityManager

Question

该程序是经常运行后台线程唤醒并将传入的DataEvents上传到远程服务器。

@Scope(value = ConfigurableBeanFactory.SCOPE_SINGLETON)
@Singleton
@Component
public class RWDataUploader {

    protected org.slf4j.Logger log = LoggerFactory.getLogger(RWDataUploader.class);

    @PersistenceContext(type = PersistenceContextType.TRANSACTION)
    private EntityManager em;

    protected EntityManager getEntityManager() {
        return em;
    }

    private final Object objectUploaderMonitor = new Object();

    private boolean isActive = true;
    private int errorCount = 0;
    private long sleepTime = 20000l;
    private Timer bgProcessTimer;

    @PostConstruct
    public void init() {
        log.info("RWDataUploadClient initilized {} ", RWDataUploader.this);
         TimerTask task = new TimerTask() {
            @Override
            public void run() {
                log.info("Staring Upload process...");
                RWDataUploader.this.startUpload();
                bgProcessTimer.cancel();
            }
        };
        bgProcessTimer = new Timer("Upload process");
        bgProcessTimer.scheduleAtFixedRate(task, 10000, 3000); // start with a delay
    }

    @PreDestroy
    public void close() {
        stopUpload();
        log.info("RWDataUploadClient closed {} ", RWDataUploader.this);
    }

    public void stopUpload() {
        isActive = false;
        synchronized (objectUploaderMonitor) {
            objectUploaderMonitor.notify();
        }
    }

    public void startUpload() {
        while (isActive) {
            synchronized (objectUploaderMonitor) {
                try {
                    uploadTxnToServer();
                    log.debug("UploadTxnToServer Done Sleeping for next cycle " + sleepTime);
                    objectUploaderMonitor.wait(sleepTime);
                } catch (Exception ex) {
                    errorCount++;
                    log.error("Exception at ObjectUploaderMonitor.wait " + ex.getMessage() , ex); 
                    if (errorCount == 1) {
                        isActive = false; 
                    }
                }
            }
        }
        log.info("Uploader client is stopped");
    }


    @Transactional
    private void uploadTxnToServer() {
        List<RWDataEvent> openDevnts = getPendingDataEvents();
        for (RWDataEvent openDevnt : openDevnts) {
            log.info("Uploaded DE {} ", openDevnt);
            // logic Upload event to remote server
            // upload done
            openDevnt.setProcessedStatusCode("UPLOADED");
            em.merge(openDevnt);
        }
    }

    private List<RWDataEvent> getPendingDataEvents() {
        Query nq = em.createNamedQuery("RWDataEvent.findByProcessedStatusCode");
        nq.setParameter("processedStatusCode", "WAIT_UPLOAD");
        return nq.getResultList();
    }

}  

执行时出现以下错误

错误 c.r.r.c.uploader.RWDataUploader - ObjectUploaderMonitor.wait 出现异常 没有可用于当前线程的实际事务的 EntityManager - 无法可靠地处理“合并”调用 javax.persistence.TransactionRequiredException：没有可用于当前线程的具有实际事务的 EntityManager - 无法可靠地处理“合并”调用 在 org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:282) 在 com.sun.proxy.$Proxy230.merge（未知来源） 在 com.retailwave.rwos.compartment.uploader.RWDataUploader.uploadTxnToServer(RWDataUploader.java:106) 在 com.retailwave.rwos.compartment.uploader.RWDataUploader.startUpload(RWDataUploader.java:82) 在 com.retailwave.rwos.compartment.uploader.RWDataUploader$1.run(RWDataUploader.java:57)

Answer 1

无需重组一切。使用TransactionTemplate 在事务中执行您的代码。

您必须使用execute 方法并将实现放入其中。