Java Threads. Пожалуйста, помогите мне запустить работу ⇐ JAVA

[Anonymous]

Прошу прощения за формулировку. Я не знаю, ясно ли я спрашиваю. Программа работает, но первый набор данных часто составляет более миллиона записей, поэтому моя программа занимает 3 дня (не фигура речи!), Чтобы запустить. (Без набора точек останова) Это работает по желанию. Первый блок кода ниже - это то, где я пытаюсь реализовать потоки. Здесь, вероятно, есть несколько проблем, так как я впервые пытаюсь реализовать поточную модель. Ответ на вопрос: «Почему ты ...» - это просто то, что я не знаю, что я здесь делаю, и я пытаюсь понять это вместе. LOL
Я ставит всю программу (меньше POJO и утилиты, которые не имеют значения) внизу. /**
* For each row, get the record from aus_national and retrieve the CRs.
*/
ExecutorService es = Executors.newCachedThreadPool();
Semaphore semaphore = new Semaphore(10);
for(CaseData currentCase : cases) {
try {
while(!es.awaitTermination(10, TimeUnit.MILLISECONDS)) {
es.execute(new WriteCR(config, currentCase.getCase_nbr(),currentCase,LOGGER,semaphore));
}
LOGGER.info("Threads Completed...");
} catch (InterruptedException e) {
LOGGER.error("Thread failed: {}",e.getMessage());
}
}

полный файл Java
package gov.usdohud.chums;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.LinkedHashSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import com.sybase.jdbc4.jdbc.SybDataSource;
import com.microsoft.sqlserver.jdbc.SQLServerDataSource;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.salientcrgt.chums.util.LogStandardOutput;

import gov.usdohud.configuration.ConfigXmlParser;
import gov.usdohud.dao.CaseData;
import gov.usdohud.helper.AUSInflator;
import gov.usdohud.model.ConfigData;

/**
* Threaded class to get CR and write files
*/
class WriteCR implements Runnable{
ConfigData config;
String caseNumber;
Logger LOGGER;
CaseData currentCase;
Semaphore semaphore;

/**
* Constructor
*/
public WriteCR (ConfigData config, String caseNumber, CaseData currentCase, Logger LOGGER, Semaphore semaphore) {
this.config = config;
this.caseNumber = caseNumber;
this.currentCase = currentCase;
this.LOGGER = LOGGER;
this.semaphore = semaphore;
}

/**
* Use case data from sfdw to get the CR(s) from aus_national.
* @param currentCase
*/
public void extractCR(Logger LOGGER, CaseData currentCase) {
String loanNumber = currentCase.getLoan_nbr();
String caseNumber = currentCase.getCase_nbr();
String scoreDate = currentCase.getScore_dt();
String aus = currentCase.getUndrwrtr_id();
String sqlString = "select file_data from aus_credit where loan_number = ? and score_dt = ? and aus = ?";

// pull CR(s)
final SQLServerDataSource dataSource = new SQLServerDataSource();
dataSource.setServerName(config.getAusNatServer().getName());
dataSource.setPortNumber(Integer.parseInt(config.getAusNatServer().getPort()));
dataSource.setDatabaseName(config.getAusNatServer().getDatabase());
dataSource.setUser(config.getAusNatServer().getUserId());
dataSource.setPassword(config.getAusNatServer().getAuthentication());
dataSource.setSendStringParametersAsUnicode(false);
try (Connection conn = dataSource.getConnection();) {
this.semaphore.acquire();
PreparedStatement stmt = conn.prepareStatement(sqlString);
stmt.setString(1, loanNumber);
stmt.setString(2, scoreDate.replaceAll("\\..*$", "").replace("-", "/"));
stmt.setString(3, aus);
ResultSet rs = stmt.executeQuery();

while (rs.next()) {
writeFile(LOGGER, caseNumber, rs.getBinaryStream("file_data"),rs.getRow());
}

} catch (SQLException e) {
LOGGER.error(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
this.semaphore.release();
}
}

public void writeFile(Logger LOGGER,String caseNumber, InputStream cr,int seq) {
final String fn = this.config.getDataFolder() + "/" + this.caseNumber + "-" + String.valueOf(seq) + ".txt";
AUSInflator decode = new AUSInflator();
byte[] crBytes = null;
try {
crBytes = cr.readAllBytes();
} catch (IOException e) {
LOGGER.error("Unable to get CR data for case {}, sequence {}", caseNumber, seq);
}
String thisCR = decode.unZipIt(crBytes).replaceAll("", "").replaceAll("", "");
try (BufferedWriter writer = new BufferedWriter(new FileWriter(fn))) {
writer.write(thisCR);
} catch (IOException e) {
LOGGER.error("Unable to create CR file for case {}, sequence {}", caseNumber, seq);
}
}

@Override
public void run() {
extractCR(LOGGER, currentCase);
}
}

/**
* Primary class with Main()
*/
public class GetCreditReports {
/** log4j logger. */
private static final Logger LOGGER = LogManager.getLogger();

/** configuration information. */
private static final ConfigData config = ConfigXmlParser.getData();

/** required to fix mail in jdk7+. */
static {
System.setProperty("java.net.preferIPv4Stack", "true");
}

public static void main(String[] args) {
// Uncomment for debugging so that the configuration objects appear in the variable stack
// ConfigData configurationValues = config;
LogStandardOutput.logSystemOutAndErrToLogger();
LOGGER.info("Start process...");

/**
* Instantiate object
*/
GetCreditReports thisRun = new GetCreditReports();

/**
* Get the data from the warehouse.
*/
LinkedHashSet cases = thisRun.getCaseData();

/**
* For each row, get the record from aus_national and retrieve the CRs.
*/
ExecutorService es = Executors.newCachedThreadPool();
Semaphore semaphore = new Semaphore(10);
for(CaseData currentCase : cases) {
try {
while(!es.awaitTermination(10, TimeUnit.MILLISECONDS)) {
es.execute(new WriteCR(config, currentCase.getCase_nbr(),currentCase,LOGGER,semaphore));
}
LOGGER.info("Threads Completed...");
} catch (InterruptedException e) {
LOGGER.error("Thread failed: {}",e.getMessage());
}
}

LOGGER.info("Process Completed...");

}

/**
* Get case data from the warehouse
* @return
*/
private LinkedHashSet getCaseData() {
LinkedHashSet cases = new LinkedHashSet();
// Query SFDW for cases
String sqlString = "select top 100 b.case_nbr, b.loan_nbr, b.score_dt, b.undrwrtr_id from idb_1 a, total_scorecard_data_final b "
+ "where a.case_nbr = b.case_nbr and endrsmnt_cy = ?";

SybDataSource datasource = new SybDataSource();
datasource.setServerName(config.getSfdwServer().getName());
datasource.setPortNumber(Integer.valueOf(config.getSfdwServer().getPort()));
datasource.setDatabaseName(config.getSfdwServer().getDatabase());
datasource.setUser(config.getSfdwServer().getUserId());
datasource.setPassword(config.getSfdwServer().getAuthentication());
try (Connection conn = datasource.getConnection();) {
PreparedStatement stmt = conn.prepareStatement(sqlString);
stmt.setLong(1, config.getFiscalYear());
ResultSet rs = stmt.executeQuery();

while (rs.next()) {
CaseData thisRow = new CaseData();
thisRow.setCase_nbr(rs.getString("case_nbr"));
thisRow.setLoan_nbr(rs.getString("loan_nbr"));
thisRow.setScore_dt(rs.getString("score_dt"));
thisRow.setUndrwrtr_id(rs.getString("undrwrtr_id"));
cases.add(thisRow);
}

LOGGER.info("Preparring to process {} scoring records.", cases.size());

} catch (SQLException e) {
LOGGER.error("SQL Exception raised: {}",e.getMessage());
}

return cases;
}

}



Подробнее здесь: https://stackoverflow.com/questions/797 ... ng-to-work