3

我正在使用 Apache TomEE 1.5.2 JAX-RS，几乎是开箱即用的，带有预定义的 HSQLDB。

下面是简化的代码。我有一个用于接收信号的 REST 样式接口：

@Stateless
@Path("signal")
public class SignalEndpoint {
    @Inject
    private SignalStore store;

    @POST
    public void post() {
        store.createSignal();
    }
}

接收到信号会触发很多事情。商店将创建一个实体，然后触发一个异步事件。

public class SignalStore {
    @PersistenceContext
    private EntityManager em;

    @EJB
    private EventDispatcher dispatcher;

    @Inject
    private Event<SignalEntity> created;

    public void createSignal() {
        SignalEntity entity = new SignalEntity();
        em.persist(entity);
        dispatcher.fire(created, entity);
    }
}

调度程序非常简单，只是为了使事件处理异步而存在。

@Stateless
public class EventDispatcher {
    @Asynchronous
    public <T> void fire(Event<T> event, T parameter) {
        event.fire(parameter);
    }
}

接收事件是另一回事，它从信号中获取数据，存储它，然后触发另一个异步事件：

@Stateless
public class DerivedDataCreator {
    @PersistenceContext
    private EntityManager em;

    @EJB
    private EventDispatcher dispatcher;

    @Inject
    private Event<DerivedDataEntity> created;

    @Asynchronous
    public void onSignalEntityCreated(@Observes SignalEntity signalEntity) {
        DerivedDataEntity entity = new DerivedDataEntity(signalEntity);
        em.persist(entity);
        dispatcher.fire(created, entity);
    }
}

对此做出反应甚至是实体创建的第三层。

总而言之，我有一个 REST 调用，它同步创建 a SignalEntity，异步触发创建 a DerivedDataEntity，异步触发创建第三种类型的实体。这一切都完美无缺，存储过程完美解耦。

除了当我以编程方式在 for 循环中触发大量（fe 1000）信号时。根据我的AsynchronousPool大小，在处理了大约一半大小的信号（相当快）后，应用程序完全冻结了几分钟。然后它恢复，以相当快的速度处理大约相同数量的信号，然后再次冻结。

AsynchronousPool在过去的半个小时里，我一直在玩设置。例如，将其设置为 2000 可以轻松地一次处理我的所有信号，而不会出现任何冻结。但在那之后，系统也不健全。触发另外 1000 个信号，导致它们被正确创建，但派生数据的整个创建从未发生过。

现在我完全不知道该怎么做。我当然可以摆脱所有这些异步事件并自己实现某种队列，但我一直认为 EE 容器的意义在于让我摆脱这种乏味。异步 EJB 事件应该已经自带了自己的队列机制。队列太满时不应立即冻结的队列。

有任何想法吗？

更新：

我现在已经用 1.6.0-SNAPSHOT 试过了。它的行为有点不同：它仍然不起作用，但我确实遇到了一个异常：

Aug 01, 2013 3:12:31 PM org.apache.openejb.core.transaction.EjbTransactionUtil handleSystemException
SEVERE: EjbTransactionUtil.handleSystemException: fail to allocate internal resource to execute the target task
javax.ejb.EJBException: fail to allocate internal resource to execute the target task
    at org.apache.openejb.async.AsynchronousPool.invoke(AsynchronousPool.java:81)
    at org.apache.openejb.core.ivm.EjbObjectProxyHandler.businessMethod(EjbObjectProxyHandler.java:240)
    at org.apache.openejb.core.ivm.EjbObjectProxyHandler._invoke(EjbObjectProxyHandler.java:86)
    at org.apache.openejb.core.ivm.BaseEjbProxyHandler.invoke(BaseEjbProxyHandler.java:303)
    at <<... my code ...>>
    ...
Caused by: java.util.concurrent.RejectedExecutionException: Timeout waiting for executor slot: waited 30 seconds
    at org.apache.openejb.util.executor.OfferRejectedExecutionHandler.rejectedExecution(OfferRejectedExecutionHandler.java:55)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
    at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:132)
    at org.apache.openejb.async.AsynchronousPool.invoke(AsynchronousPool.java:75)
    ... 38 more

就好像 TomEE 不会对操作进行任何排队。如果在调用的那一刻没有线程可以自由处理，那么运气不好。当然，这不是故意的..？

更新 2：

好的，我似乎偶然发现了一个半解决方案：将AsynchronousPool.QueueSize属性设置为 maxint 可以解决冻结问题。但问题仍然存在：为什么 QueueSize 首先如此有限，更令人担忧的是：为什么这会阻塞整个应用程序？如果队列已满，它会阻塞，但是一旦从中取出一个任务，另一个应该会弹出，对吗？队列似乎被阻塞，直到它再次完全为空。

更新 3：

对于任何想要尝试的人：http: //github.com/JanDoerrenhaus/tomeefreezetestcase

更新 4：

事实证明，增加队列大小并不能解决问题，它只会延迟它。问题仍然存在：一次异步操作太多，并且 TomEE 阻塞非常严重，以至于它甚至无法在终止时取消部署应用程序。

到目前为止，我的诊断是任务清理无法正常工作。我的任务都非常小而且很快（参见github 上的测试用例）。我已经担心 OpenJPA 或 HSQLDB 会因为太多并发调用而减慢速度，但是我注释掉了所有em.persist调用，问题仍然存在。因此，如果我的任务非常小而且速度很快，但仍然设法阻止 TomEE 如此糟糕以至于在 30 秒后无法获得任何进一步的任务（javax.ejb.EJBException: fail to allocate internal resource to execute the target task），我会想象完成的任务会徘徊，堵塞管道，可以这么说.

我该如何解决这个问题？

4

2 回答 2

2

基本上 BlockingQueues 使用锁来确保数据的一致性并避免数据丢失，因此在过高并发的环境中它会拒绝很多任务（您的情况）。

您可以使用 RejectedExecutionHandler 实现在中继上播放以重试提供任务。一种实现可以是：

new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
            for (int i = 0; i < 10; i++) {
                if (executor.getQueue().offer(r)) {
                    return;
                }

                try {
                    Thread.sleep(50);
                } catch (final InterruptedException e) {
                    // no-op
                }
            }
            throw new RejectedExecutionException();
        }
    }

它甚至更适合随机睡眠（在最小值和最大值之间）。

这个想法基本上是：如果队列已满，请稍等片刻以降低并发性。

于 2013-08-05T15:44:18.307 回答
1

通过 WEB-INF/application.properties https://issues.apache.org/jira/browse/TOMEE-1012进行配置

于 2014-02-16T07:36:42.460 回答