如何处理CDI创建的Runnable

Question

我有一个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);
    }
  }
}

这是我的Scheduler类的代码

@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，它正在使用注入它的Object的Scope;这将使它成为@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？

Answer 1

所以我似乎错误地使用了@ThreadScoped注释。它不是在Runnable类上，而是在注入bean的生产者上。使用我更新的代码，注入的EntityManager和服务正在被使用，但是在Weld容器关闭之前，Runnable本身不会被释放。

这是我的制片人

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