TA貢獻(xiàn)1818條經(jīng)驗(yàn) 獲得超11個(gè)贊
我有一些課：
@Component
public MyClass {
private volatile boolean stopped = false;
public void verification() throws Exception {
Thread kpiAllThread = getKPIAllThread();
try {
for (int i = 0; i < poolSize; i++) {
execDispatcher.put(processExecutor.submit(getCheckValuesInKPIConsumerTask(workingQueue)));
}
kpiAllThread.start();
} finally {
waitFinished();
}
}
public void setStop(bolean stopped) {
this.stopped = stopped;
}
private Thread getKPIAllThread() {
return new Thread(() -> {
try {
LOG.debug("KPIAllThread started!");
dao.getKpiAll(workingQueue);
for (int i = 0; i < poolSize; i++) {
workingQueue.put(() -> true);
}
} catch (Exception ex) {
LOG.error("KPIAllThread exception: ", ex);
} finally {
LOG.error("KPIAllThread finished!");
}
});
}
}
這個(gè)類啟動(dòng)生產(chǎn)者線程getKPIAllThread。他從 db 獲取數(shù)據(jù)并放入BlockingQueue.
getKpiAll像這樣的方法：
public void getKpiAll(final BlockingQueue<KeyPropertyIndex> kpiData) throws Exception {
LOG.debug("Starting getKpiAll");
try (final Connection con = dataSource.getConnection();
final Statement stmt = con.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
stmt.setFetchSize(Integer.MIN_VALUE);
try (final ResultSet rs = stmt.executeQuery(sqlGetAllkpi)) {
while (rs.next()) {
kpiData.put(new KeyPropertyIndexData(rs.getLong(1), rs.getString(2)));
}
}
LOG.debug("Finished get getKpiAll");
} catch (Exception ex) {
throw ex;
}
}
還有一個(gè)變量stopped可以從外部設(shè)置到true. 這樣做時(shí)如何安全地停止我的線程？這樣到數(shù)據(jù)庫的所有連接都關(guān)閉并且線程成功完成？
舉報(bào)