0

我有一个 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？

4

1 回答 1

0

所以看来我一直在错误地使用@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
于 2019-03-14T14:10:17.220 回答