Spring调度器和线程池执行器

Question

我需要帮助设计基于多线程的应用程序，包括动态url创建和线程处理。

我在我的应用程序中使用了一个Spring调度器，它每30秒调度一次。从这个调度方法中，我调用了一些基于服务的api，它在循环中，而且我需要每个API有一个线程池执行器，上面有一个线程处理。

由于这个过程是从计划方法开始的，所以每次创建新的线程池时，这就是问题所在。你可以在代码中看到。

我想要的是，如果对于任何一个应用编程接口，如果线程池已经存在，那么该线程应该能够通过不创建新的线程池来自己识别并开始处理。

让我知道，如果需要更多的信息。

欢迎任何建议，谢谢

#API Properties
service.url=http://{0}.abc.net/xyz.php?
service.urls = abc1, abc2, abc3, abc4, abc5

@Service
public class APIServiceImpl implements APIService {

@Autowired
MsgService msgService;

private static final Logger LOGGER = Logger.getLogger(APIServiceImpl.class);
private static Properties fileProperties = PropertyUtility.getfileProperties();

@Scheduled(fixedDelayString = "30000")
public void getServiceMessage() throws ServiceException {
    try {
        long startTime = System.currentTimeMillis() / 1000L;
        long endTime = startTime + 30;
        String urlStr = fileProperties.getProperty("service.urls");
        String[] urls = urlStr.split(",");
        foreach(String url : urls){
            serviceApi(url.trim(), startTime, endTime);
        }
    } catch (Exception e) {
        throw new ServiceException(e);
    }
}

private void serviceApi(String url, long startTime, long endTime) {
    StringBuffer buffer = new StringBuffer();
    buffer.append(java.text.MessageFormat.format(fileProperties.getProperty("service.url"), url));
    buffer.append("starttime=" + startTime);
    buffer.append("&endtime=" + endTime);
    ExecutorService executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());
    executor.submit(new APIThreadHandler(buffer.toString(), messageService));
}  

public class APIThreadHandler implements Runnable {

private static final Logger LOGGER = Logger.getLogger(APIThreadHandler.class);
private String url;
MsgService msgService;

public APIThreadHandler(String url, MsgService msgService) {
    this.url = url;
    this.msgService = msgService;
}

public void run() {
    System.out.println("ThreadID: " + Thread.currentThread().getId());
    try {
        URL srcUrl = new URL(url);
        List<Map<?, ?>> data = readObjectsFromCsv(srcUrl);
        JSONArray jsonArray = new JSONArray(data);
        for (int i = 0; i < jsonArray.length(); i++) {
            msgService.sendMessage(jsonArray.getJSONObject(i));
        }
    } catch (Exception e) {
        LOGGER.error("Exception occured - ", e);
    }
}

private static List<Map<?, ?>> readObjectsFromCsv(URL url) {
    List<Map<?, ?>> listMap = null;
    try {
        CsvSchema bootstrap = CsvSchema.emptySchema().withHeader();
        CsvMapper csvMapper = new CsvMapper();
        MappingIterator<Map<?, ?>> mappingIterator = csvMapper.readerFor(Map.class).with(bootstrap).readValues(url);
        listMap = mappingIterator.readAll();
    } catch (Exception e) {
        LOGGER.error("Exception occured - ", e);
    }
    return listMap;
}

Answer 1

您可以使用@PostConstruct初始化线程池一次，每次都可以重复使用。当应用程序上下文被加载时，它将只被调用一次。

@PostConstruct
public void init() {
 executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());
}