javaee使用timerservice的多会话bean

zynd9foi  于 2021-07-05  发布在  Java
关注(0)|答案(1)|浏览(388)

我有一个带有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替代方案允许访问引用的资源,并且在示例化之后只调用一次?
提前感谢您提供任何建设性的信息。

uurity8g

uurity8g1#

我自己找到了一个解决方案,所以我把它贴在这里,这样可以帮助别人。
我添加了一个新的单例启动bean,由于cdi,它保存了实现myjobexecutor接口的所有bean的列表。在jee环境中,cdi与ejb配合得很好,因此它注入了会话bean!警告:cdi将只注入直接实现myjobexecutor的类,因此如果您有一个实现myjobexecutor的抽象类和一个从此抽象类扩展的具体bean,cdi将不起作用。它必须明确 implements MyJobExecutor . 在startup类的postconstruct中,我调用每个myjobexecutorbean在其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() 价值观。但是,当前一个超时仍在运行时,timerservice不会触发超时。这将阻止预期的并行处理:(
为了解决这个问题,我调整了timeout方法来不执行业务逻辑本身,而是启动一个执行业务逻辑的托管线程。这条路 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接口的所有会话bean在部署时自动注册第一个计时器。
可以注册多个定时器与同一时间,他们将并行运行。
通过使用managedthreadfactory,线程还可以访问jndi上下文并可以查找EJB。

相关问题