javaee使用timerservice的多会话bean

zynd9foi 于 2021-07-05 发布在 Java

关注(0)|答案(1)|浏览(391)

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

Java jakarta-ee weblogic12c

来源：https://stackoverflow.com/questions/63742080/java-ee-multiple-sessionbeans-with-timerservice

1条答案

按热度按时间

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。

赞(0）回复(0）举报 2021-07-05

我来回答

javaee使用timerservice的多会话bean

1条答案

相关问题

热门标签

最新问答