带有TimerService的Java EE多个SessionBeans

Question

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

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

通过这种方式，我可以控制何时重新启动，而且由于TimerService回调，我没有Hibernate线程。

我希望多次出现此会话bean,以预测队列上的瓶颈：

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

@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
        }
    }
}

问题：错误出现在@postconstruct带注释的init()方法中：此时不允许使用TimerService。当我将sessionbean设为@singleton时，这是允许的，但随后我就失去了并行处理队列的可能性。有没有人有办法解决这个问题？如果TimerService不是正确的机制，还有什么可供选择的。是否有一个PostConstruct替代方案，允许访问引用的资源，并且在实例化之后只调用一次？

提前感谢任何建设性的信息。

Answer 1

我自己找到了一个解决方案，所以我把它贴在这里，这样它可能会帮助其他人。

我添加了一个新的单例Startup bean，它包含一个列表，其中列出了通过CDI实现MyJobExecutor接口的所有bean。在JEE环境中，CDI可以很好地与EJB一起工作，因此它可以注入会话bean！警告：CDI只会注入直接实现MyJobExecutor的类，因此如果您有一个实现MyJobExecutor的抽象类和一个从这个抽象类扩展的具体bean，它就不起作用了。它必须显式地实现MyJobExecutor。在startup类的postconstruct中，我调用每个MyJobExecutor bean在其TimerService中注册一个新的计时器。这样，我就可以为每个会话bean创建第一个singleActionTimer（s）。

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不会触发超时。这将阻止预期的并行处理：（

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接口的所有会话bean自动注册第一个计时器。
• 注册多个具有相同时间的计时器的可能性，并且它们将并行运行。
• 通过使用ManagedThreadFactory，线程还可以访问JNDI上下文并查找EJB。