EJB中的线程.sleep()

Question

我知道在EJB里面搞线程是大忌，但我想请教一下如何处理这个案例。我的EJB正在调用一个外部的Web服务，这个服务有时会返回一个 "繁忙 "状态。当这种情况发生时，我希望等待一段时间，然后使用与之前相同的数据重新提交请求。

实现这个问题的最好方法是什么？

Answer 1

中的"..."。EJB限制常见问题 它明确指出，你

不应创建或管理线程

而让一个线程进入睡眠状态也算作 "管理 "它。

在你的例子中，当web服务返回 "忙 "的状态时，你可以安排一个作业在稍后的时间点重试发送消息，例如通过使用 石英调度器. 执行将就此结束，任何进一步的处理都应该委托给作业调度器。

Answer 2

EJB 3.1带来了一个新的 @Asynchronous 特征 你可以利用的。

@Asynchronous
@TransactionAttribute(NOT_SUPPORTED)
public Future<WebServiceResult> callWebService(int retries) {
    WebServiceResult result = webService.call();

    if (!result.equals(BUSY)) {
        return result;
    }

    if (retries <= 0) {
        throw new TooBusyException();
    }

    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }

    return callWebService(retries - 1);
}

然后简单地调用你的网络服务，

Future<WebServiceResult> result = yourEJB.callWebService(1);

// Can do some interesting stuff here.
// ...
// ...

result.get(2, SECONDS);  // Block for up to 2 seconds.

如你所见，你可以免费获得可配置的重试次数和超时次数。

这与仅仅调用 Thread.sleep()? 返回 Future 是比较明确和容易管理的。另外，我不认为 Thread.sleep() 是那么的有害。唯一的问题是，现在这个EJB实例不能再被其他客户重用了。有了 Future 异步调用发生在其他EJB和线程池内。至于 Thread#interrupt() 挡板内，请参考 为什么在捕捉任何InterruptException时要调用Thread.currentThread.interrupt()？

另一个想法：在调用web服务时使用方面，捕捉 BusyException 一次，然后重试。

Answer 3

但是外部的web服务是外部的，你要给它开通网络连接，你希望做一些管理的事情。 这就是JCA的作用，而不是EJB。

Answer 4

这里有一个建议(或者说是另一种方法)，用简单控件的池子。

1 - 把你的上下文(EJB)作为参数传给你的方法(rest endpoint, scheduler, default's methods)

2-用互补的调度器或实体标志来控制状态。

3 - 小心数据处理的数量

4-建议。强烈建议指标、日志和测试、试验、测试。

5 - 这段代码是在SpringBoot上，但在Jboss中进行了测试(经过修改)，在EJB上下文下--仔细测试。

6-根据您的意愿进行修改。(发送建议-评论)

BaseControlExecutor.java

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;


public class BaseControlExecutor {

    private final ScheduledThreadPoolExecutor poolExec = new ScheduledThreadPoolExecutor(2);

    public void execWithTimeout(final Runnable runnable, long timeout,
            TimeUnit timeUnit) throws Exception {
        execWithTimeout(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                runnable.run();
                return null;
            }
        }, timeout, timeUnit);
    }

    public <T> T execWithTimeout(Callable<T> callable, long timeout,    TimeUnit timeUnit) throws Exception {

        final Future<T> future = poolExec.submit(callable);

        try {
            return future.get(timeout, timeUnit);
        } catch (TimeoutException e) {
            future.cancel(true);
            throw e;
        } catch (ExecutionException e) {
            Throwable t = e.getCause();
            if (t instanceof Error) {
                throw (Error) t;
            } else if (t instanceof Exception) {
                throw (Exception) t;
            } else {
                throw new IllegalStateException(t);
            }
        }
    }
}

EndpointControlRest.java

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;

@RestController
@RequestMapping(value = "/report")
@Api(tags = "Endpoint of Future")
public class EndpointControlRest extends BaseControlExecutor {

    Logger logger = LoggerFactory.getLogger(EndpointControlRest.class);

    //single metric of execution
    protected final AtomicLong counter = new AtomicLong();

    @GetMapping(path = "/withThread", produces = { "application/json" })
    @ApiOperation(value = "Return Hello count.")
    public String greeting() {

        Long countRunner = counter.incrementAndGet();
        String json = ""; //or EJB context to use in Thread - becareful

        new Thread(() -> {

            try {
                execWithTimeout(new Runnable() {
                    @Override
                    public void run() {

                        Instant start = Instant.now();
                        logger.info("Report init - " + countRunner);

                        //generating reports
                        generateBackgroundReport(json);

                        logger.info("Report End - " + countRunner);

                        Instant finish = Instant.now();
                        long timeElapsed = Duration.between(start, finish).toMillis();

                        logger.info("###DEBUG - " + countRunner + " - OK |Time exe: " + timeElapsed);

                    }
                }, 120, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                logger.info("###DEBUG - " + countRunner + " - Timeout - " + e.getMessage());
            } catch (Exception e) {
                logger.info("###DEBUG - " + countRunner + " - Exception - " + e.getMessage());
            }
        }).start();

        logger.info("####DEBUG - Rest call released");
        return "Hello " + countRunner;
    }

    public String generateBackgroundReport(String json){

        //simulating work
        Long x = 0L;
        for(Long i = 0L; i < 1000000000L; i ++){
            x = i + 1;
        }
        logger.info("####DEBUG -report: " + x);
        return "OK";
    }
}