java - 如何处理 CDI 创建的 Runnable
问题描述
我有一个 Java SE 项目,它接受一些命令行参数,在单独的线程中为指定的参数执行处理。我正在使用以下焊接微配置文件配置依赖项进行注入
<dependency>
<groupId>org.jboss.weld.se</groupId>
<artifactId>weld-se-core</artifactId>
<version>3.1.0.Final</version>
</dependency>
<dependency>
<groupId>org.wildfly</groupId>
<artifactId>wildfly-microprofile-config-implementation</artifactId>
<version>1.2.1</version>
</dependency>
这是我的 beans.xml
<?xml version="1.0"?>
<beans xmlns="http://xmlns.jcp.org/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/beans_1_1.xsd"
version="1.1" bean-discovery-mode="all">
<decorators>
<class>org.jboss.weld.environment.se.threading.RunnableDecorator</class>
</decorators>
</beans>
我的项目从以下主类开始
@ApplicationScoped
public class Main {
@Inject
private Scheduler scheduler;
public void process(List<String> types) throws InterruptedException {
scheduler.schedule(types);
}
public static void main(String[] args) throws InterruptedException {
SeContainerInitializer initializer = SeContainerInitializer.newInstance();
try (SeContainer container = initializer.initialize()) {
Main main = container.select(Main.class).get();
List<String> argsList = Arrays.asList(args);
final List<String> types = parseArguments(argsList);
main.process(types);
}
}
}
这是我的调度程序类的代码
@ApplicationScoped
public class Scheduler {
private static final Duration DEFAULT_WAIT_TIME = Duration.ofSeconds(30);
@Inject
@ConfigProperty(name = "POOL_SIZE", defaultValue = "10")
@Getter
private int poolSize = 5;
@Inject
@ConfigProperty(name = "WAIT_DURATION", defaultValue = "PT30S")
@Getter
private String durationStr;
@Getter
private Duration waitDuration;
private ThreadPoolExecutor executor;
@Inject
private Instance<ExportRunner> exports;
@PostConstruct
public void init() {
executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolSize);
try {
waitDuration = Duration.parse(durationStr);
} catch (DateTimeException | NullPointerException e) {
waitDuration = DEFAULT_WAIT_TIME;
}
}
public void schedule(Collection<String> types) throws InterruptedException {
if (types != null && !types.isEmpty()) {
//Spawn a new thread for each type
for(String type : types) {
ExportRunner runner = exports.get();
runner.setType(type);
executor.submit(runner);
}
} else {
throw new IllegalArgumentException("No FileTypes provided. Not performing export");
}
//Wait until every thread has completed
while(getTotalThreads() > 0) {
Thread.sleep(waitDuration.toMillis());
}
//shutdown executor which effectively ends the program
executor.shutdownNow();
}
public int getTotalThreads() {
return getActiveCount() + getQueueSize();
}
public int getActiveCount() {
return executor.getActiveCount();
}
public int getQueueSize() {
return executor.getQueue().size();
}
}
这是 Runnable 的骨架
@Dependent
public class ExportRunner implements Runnable {
@Setter
private FileType type;
//Has a custom producer and disposer
@Inject
@SEDataSource
private EntityManager em;
//Has a custom producer and disposer
@Inject
@SEDataSource
AlertService alertService;
//Has a custom producer and disposer
@Inject
@SEDataSource
HistoryService historyService;
@PostConstruct
private void init() {
//Set to same entity manager so that
//everythings happen inside single transaction
alertService.setEm(em);
historyService.setEm(em);
}
@PreDestroy
public void cleanup() {
log.info("ExporterRunner @PreDestroy was called");
}
public void run() {
try {
//do processing
} finally {
log.info("Processing Complete");
}
}
}
我遇到的问题是注入的对象(Runnable 和包含的服务和 EntityManager)并且在所有线程完成并执行 executor.shutdown() 命令之前永远不会被释放。
我相信由于 Runner 被标记为@Dependent,它使用的是注入它的对象的作用域;这将使它成为@ApplicationScoped。我尝试使用@ThreadScoped (org.jboss.weld.environment.se.contexts.ThreadScoped) 标记该类,如下所示
@ThreadScoped
public class ExportRunner implements Runnable {
...
}
但这会导致以下异常
org.jboss.weld.contexts.ContextNotActiveException: WELD-001303: No active contexts for scope type org.jboss.weld.environment.se.contexts.ThreadScoped
我觉得我需要使用@ActivateThreadScope (org.jboss.weld.environment.se.contexts.activators.ActivateThreadScope) 注释,但我还没有找到任何如何使用它的示例。有谁知道我如何使我的 Runnable 不是 @ApplicationScoped?
解决方案
所以看来我一直在错误地使用@ThreadScoped 注释。它不在 Runnable 类上,而是在注入 bean 的生产者上。使用我更新的代码,注入的 EntityManager 和服务将被处理掉,但 Runnable 本身直到 Weld 容器关闭后才会被处理掉。
这是我的制片人
public class ProjectProducer {
//Producer needs to have the @ThreadScopes
@Produces
@SEDataSource
@ThreadScoped
EntityManager openEntityManager() {
log.info("Creating Entity manager");
...
}
//Note that @ThreadScoped is not required for Disposer
public void closeEntityManager(@Disposes @SEDataSource EntityManager em) {
log.info("Disposing of EntityManager");
}
//Producer needs to have the @ThreadScopes
@Produces
@ThreadScoped
@SEDataSource
public AlertService createAlertService() {
log.info("Creating Alert Service");
...
}
//Note that @ThreadScoped is not required for Disposer
public void disposeAlertService(@Disposes @SEDataSource AlertService alertService) {
log.info("Disposing AlertService");
}
//Producer needs to have the @ThreadScopes
@Produces
@ThreadScoped
@SEDataSource
public FileHistoryService createFileHistoryService() {
log.info("Creating History Service");
}
//Note that @ThreadScoped is not required for Disposer
public void disposeFileHistoryService(@Disposes @SEDataSource FileHistoryService service) {
log.info("Disposing FileHistoryService");
}
}
Runnable 类没有太大变化,但我正在展示 CDI 相关部分
public class ExportRunner implements Runnable {
//The injection point remained the same
@Inject
@SEDataSource
private EntityManager em;
@Inject
@SEDataSource
AlertService alertService;
@Inject
@SEDataSource
FileHistoryService historyService;
//This should be an @PostConstruct, but because ThreadScoped context does
//not exist until after the run method is called it must be called inside it
public void init() {
alertService.setEm(em);
historyService.setEm(em);
}
//actual PostConstruct to show when Runnable was created
@PostConstruct
public void postConstruct() {
log.info("Creating ExportRunner");
}
//PreDestory method to show that the Thread was being disposed
@PreDestroy
public void preDestory() {
log.info("ExporterRunner @PreDestroy was called");
}
public void run() {
try {
log.info("Starting run method");
//Force initialization since we can't use @PostConstruct
init();
//do processing
} finally {
log.info("Processing Complete");
}
}
}
下面是显示注入和清理顺序的日志输出
Creating ExportRunner
Starting run method
Creating Alert Service
Creating History Service
Processing Complete
Disposing FileHistoryService
Disposing AlertService
Disposing of EntityManager
ExporterRunner @PreDestroy was called
推荐阅读
- php - 如何在数据库中选择特定用户并仅显示他的信息?
- visual-studio-code - 没有与“启用自定义 CSS 和 JS”匹配的命令
- powershell - 将 CSV 文件导入现有 Excel 工作簿中的新 Excel 工作表
- c# - 我如何知道队列何时结束我的线程消耗了 MQ 消息
- python - 对包含字符串的熊猫数据框中的列进行求和
- time-series - Druid Timeseries 行计数聚合
- vue.js - 如何在方法中使用 Vuex 状态数据
- php - Apache 在启动库未加载时给出错误原因:即使文件存在,也找不到图像
- c# - NavigationView.Loaded 方法没有被调用
- documentum - 即使没有添加到restricted_folder_ids 中,是否可以向用户显示其他文件柜?