具有TimerService的Java EE多个SessionBean

问题描述

我有一个带有TimerService的无状态会话Bean。超时时，它将开始消耗JMS队列。在处理消息时，它需要访问可能是暂时不可用的外部资源。超时方法会循环调用MessageConsumer.receiveNowait()，直到：

没有更多消息可处理：它注册了一个新计时器=现在+ 10分钟。结束。
在处理过程中发生错误：它回滚消息并注册一个新计时器：现在+ 30分钟。结束。

这样，我可以控制何时重新启动，并且由于TimerService回调而没有睡眠线程。

我希望该会话bean多次出现，以预测队列中的瓶颈：

                    +-----<ejb>-------+
                    | timerService    |
                    |                 |              +---------------------+
                ----| onTimeout() {}  | -----------> | external dependency |
               /    |                 |         /    +---------------------+ 
              /     +-----------------+        /
             /                                /
+---------+ /                                /
|||queue|||K                                /
+---------+ \                              /
             \      +-----<ejb>-------+   /
              \     | timerService    |  /
               \    |                 | /
                ----| onTimeout() {}  | 
                    |                 |
                    +-----------------+

我的会话bean看起来像这样（当然是简化的）：

@Stateless
public class MyJob {
    
    @Resource
    private TimerService timerService;

    @postconstruct
    public void init() {
        registerNewTimer(1000L); // -> problem: timerService not accessible
        System.out.println("Initial Timer created");
    }

    private void registerNewTimer(long duration) {
        TimerConfig config = new TimerConfig();
        config.setPersistent(false);
        timerService.createSingleActionTimer(duration,config);
    }

    @Timeout
    @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
    public void execute() {
        try {
            // instantiate JMS session and consumer
            while ((message = consumer.receiveNowait()) != null) {
                // business logic with message
                message.ackNowledge();
            }
            // queue empty,let's stop for Now and register a new timer + 10min
            registerNewTimer(10*60*1000);
        } catch (ResourceException re) {
            // external resource unavailable,let's wait 30min
            registerNewTimer(30*60*1000);
            // last message not ackNowledged,so rolled back
        }
    }
}

我不想使用消息驱动Bean，因为我想控制何时使用消息（请参见发生错误时的延迟逻辑）。

问题：错误发生在带有@postconstruct注释的init()方法中：目前，不允许使用timerService。当我使sessionbean @Singleton时允许使用，但是后来我失去了并行处理队列的可能性。有谁知道如何解决这个问题？如果TimerService不是正确的机制，则可以选择替代方法。是否存在postconstruct替代方案，该替代方案允许访问引用的资源并且在实例化后仅被调用一次？

在此先感谢您提供任何有益的信息。

解决方法

我自己找到了一个解决方案，所以我将其发布在这里，以便对他人有所帮助。

由于CDI，我添加了一个新的Singleton，Startup bean，其中包含实现MyJobExecutor接口的所有bean的列表。在JEE环境中，CDI与EJB配合良好，因此可以注入会话bean！警告：CDI将仅注入直接实现MyJobExecutor的类，因此，如果您有一个实现MyJobExecutor的抽象类以及从该抽象类扩展的具体bean，则CDI将不起作用。它必须显式implements MyJobExecutor。在启动类的后构建中，我调用每个MyJobExecutor bean在其TimerService中注册一个新计时器。这样，我可以为每个会话bean创建第一个singleActionTimer。

public interface MyJobExecutor {
    // how many timers to register for this bean meaning how many parallel processes.
    public int nrOfParallellInstances();
}

import javax.annotation.PostConstruct;
import javax.ejb.EJB;
import javax.ejb.Singleton;
import javax.ejb.Startup;
import javax.enterprise.inject.Any;
import javax.enterprise.inject.Instance;
import javax.inject.Inject;

@Singleton
@Startup
public class MyJobStartup {

    @Inject
    @Any
    private Instance<MyJobExecutor> myJobs;
    
    @PostConstruct
    public void startIt() {
        for (MyJobExecutor ajob : myJobs) {
            for(int i=0; i<ajob.nrOfParallellInstances(); i++)
                // register every instance with a 1 second delay between each timeout
                ajob.registerNewTimer(1000L*(i+1));
        }
    }
}

内部循环将基于ajob.nrOfParallellInstances()值为同一个bean注册多个计时器。但是，当上一个超时仍在运行时，TimerService将不会触发超时。如预期的那样，这将阻止并行处理：（

为解决此问题，我对超时方法进行了调整，使其不执行业务逻辑本身，而是启动一个执行业务逻辑的托管线程。这样，onTimeout方法将快速结束，计时器服务将启动下一个超时，从而导致多个并行执行（每个并行执行在托管线程中）：

import javax.enterprise.concurrent.ManagedThreadFactory;
import javax.naming.InitialContext;

@Resource(name = "concurrent/AsyncJobThreadFactory")
private ManagedThreadFactory threadFactory;

@Timeout
private void onTimeout(Timer timer) {
    LOG.info("Timeout triggered for " + timer.getInfo());
    Thread thread = threadFactory.newThread(new AsyncExecRunnable((String) timer.getInfo()));
    if (thread != null) {
        thread.start();
    } else {
        LOG.warn("No thread available for this job. Retry in 5 minutes.");
        registerNewTimer(1000L * 60 * 5);
    }

}
private static class AsyncExecRunnable implements Runnable {

    private String extra info;
    
    public AsyncExecRunnable(String info) {
        this.info = info;
    }

    @Override
    public void run() {
        try {
            LOG.info("Job executor thread started for " + info);
            InitialContext ic=new InitialContext();
            // business logic. With ic you can lookup other EJBs
        } catch (Exception e) {
            LOG.error("Problem executing AsyncJobExecutor:",e);
        }
    }
}

通过此设置，我可以：

在部署时为实现MyJobExecutor接口的所有Session bean自动注册首次计时器。
可以同时注册多个计时器，它们将并行运行。
通过使用ManagedThreadFactory，线程也可以访问JNDI上下文并可以查找EJB。

jakarta-ee java java weblogic12c