ThreadLocal、マルチスレッド、CountDownLatch使用例
13482 ワード
@Service
@Slf4j
public class WeatherServiceImpl implements IWeatherService {
@Resource
private WeatherServiceApi weatherServiceApi;
@Resource
private RedisTemplate redisTemplate;
@Value("${redis.weather}")
private Integer redisWeather;
private ThreadLocal hourlyThreadLocal = new ThreadLocal();
private ThreadLocal dailyThreadLocal = new ThreadLocal<>();
@Override
public WeatherHourlyAndLifestyleNow queryCurrentDayWeather(String location,String lifestyle) {
Long st = System.currentTimeMillis();
String weatherKey = RedisEnum.CpspWeather.LOCATION_LIFESTYLE_SAVE_HOURLY_WEATHER.getKey(location,lifestyle);
WeatherHourlyAndLifestyleNow data = (WeatherHourlyAndLifestyleNow)redisTemplate.opsForValue().get(weatherKey);
if (null != data){
return data;
}
//data = new WeatherHourlyAndLifestyleNow();
OnTheDayHourlyWeatherThreadLocal onTheDayHourlyWeatherThreadLocal = new OnTheDayHourlyWeatherThreadLocal(location,weatherServiceApi,redisTemplate,redisWeather,null,null,true);
hourlyThreadLocal.set(onTheDayHourlyWeatherThreadLocal.queryCurrentDayWeather(location, lifestyle));
data = hourlyThreadLocal.get();
// CountDownLatch countDownLatch = new CountDownLatch(hourlyCountDownLatch);
// FutureTask weatherHourlyTask = new FutureTask(new WeatherHourlyTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather));
// FutureTask lifestyleTask = new FutureTask(new LifestyleTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather,lifestyle));
// FutureTask weatherForecastTask = new FutureTask(new WeatherForecastTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather,null,null,true));
// FutureTask airQualityTask = new FutureTask(new AirQualityTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather,null,null,true));
// ExecutorService executorService = Executors.newFixedThreadPool(hourlyCountDownLatch);
// executorService.submit(weatherHourlyTask);
// executorService.submit(lifestyleTask);
// executorService.submit(weatherForecastTask);
// executorService.submit(airQualityTask);
// Long ft = System.currentTimeMillis();
// log.debug("==> ft run consume time :{}, countDownLatch :{}",ft-st,countDownLatch);
// try {
// countDownLatch.await();
// data.setHourlyList((List) weatherHourlyTask.get());
// data.setLifestyleNows((List) lifestyleTask.get());
// WeatherDailyForecastVo dailyForecastVo = (WeatherDailyForecastVo)weatherForecastTask.get();
// if (null != dailyForecastVo){
// data.setDailyForecast(dailyForecastVo.getDailyForecastList().get(0));
// data.setBaseInfoVo(dailyForecastVo.getBaseInfoVo());
// }
// List airQualityList = (List)airQualityTask.get();
// if(!CollectionUtils.isEmpty(airQualityList)){
// data.setAirQuality(airQualityList.get(0));
// }
// } catch (InterruptedException | ExecutionException e) {
// log.error("error:",e);
// throw ExceptionEnum.QUERY_WEATHER_ERROR.getException();
// }finally {
// executorService.shutdown();
// }
if (null == data){
redisTemplate.opsForValue().set(weatherKey,data,redisWeather,TimeUnit.SECONDS);
}
log.debug("");
Long ed = System.currentTimeMillis();
log.debug("===> weatherServiceImpl queryCurrentDayWeather consume time total : {}",ed - st);
log.debug("");
return data;
}
}
@Slf4j
public class OnTheDayHourlyWeatherThreadLocal {
private String location;
private WeatherServiceApi weatherServiceApi;
private RedisTemplate redisTemplate;
private Integer redisWeather;
private String startDate;
private String endDate;
private Boolean onTheDay;
private int hourlyCountDownLatch = 4;
private static ThreadLocal threadLocal = new ThreadLocal();
private CountDownLatch countDownLatch = new CountDownLatch(4);
public OnTheDayHourlyWeatherThreadLocal(String location, WeatherServiceApi weatherServiceApi,
RedisTemplate redisTemplate, Integer redisWeather, String startDate, String endDate,
Boolean onTheDay){
this.location = location;
this.weatherServiceApi = weatherServiceApi;
this.redisTemplate = redisTemplate;
this.redisWeather = redisWeather;
this.startDate = startDate;
this.endDate = endDate;
this.onTheDay = onTheDay;
}
public WeatherHourlyAndLifestyleNow queryCurrentDayWeather(String location, String lifestyle) {
Long st = System.currentTimeMillis();
String weatherKey = RedisEnum.CpspWeather.LOCATION_LIFESTYLE_SAVE_HOURLY_WEATHER.getKey(location,lifestyle);
WeatherHourlyAndLifestyleNow data = (WeatherHourlyAndLifestyleNow)redisTemplate.opsForValue().get(weatherKey);
if (null != data){
return data;
}
data = new WeatherHourlyAndLifestyleNow();
FutureTask weatherHourlyTask = new FutureTask(new WeatherHourlyTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather));
FutureTask lifestyleTask = new FutureTask(new LifestyleTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather,lifestyle));
FutureTask weatherForecastTask = new FutureTask(new WeatherForecastTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather,null,null,true));
FutureTask airQualityTask = new FutureTask(new AirQualityTask(countDownLatch,location,weatherServiceApi,redisTemplate,redisWeather,null,null,true));
ExecutorService executorService = Executors.newFixedThreadPool(hourlyCountDownLatch);
executorService.submit(weatherHourlyTask);
executorService.submit(lifestyleTask);
executorService.submit(weatherForecastTask);
executorService.submit(airQualityTask);
try {
countDownLatch.await(3,TimeUnit.SECONDS);
data.setHourlyList((List) weatherHourlyTask.get());
data.setLifestyleNows((List) lifestyleTask.get());
WeatherDailyForecastVo dailyForecastVo = (WeatherDailyForecastVo)weatherForecastTask.get();
if (null != dailyForecastVo){
data.setDailyForecast(dailyForecastVo.getDailyForecastList().get(0));
data.setBaseInfoVo(dailyForecastVo.getBaseInfoVo());
}
List airQualityList = (List)airQualityTask.get();
if(!CollectionUtils.isEmpty(airQualityList)){
data.setAirQuality(airQualityList.get(0));
}
} catch (InterruptedException | ExecutionException e) {
log.error("error:",e);
throw ExceptionEnum.QUERY_WEATHER_ERROR.getException();
}finally {
executorService.shutdown();
}
Long ed = System.currentTimeMillis();
log.debug("===> WeatherHourlyThreadLocal queryCurrentDayWeather consume time total : {}",ed - st);
log.debug("");
return data;
}
}
その中の一つFutureTask
@Slf4j
public class AirQualityTask implements Callable> {
private CountDownLatch countDownLatch;
private String location;
private WeatherServiceApi weatherServiceApi;
private RedisTemplate redisTemplate;
private Integer redisWeather;
private String startDate;
private String endDate;
private Boolean onTheDay;
public AirQualityTask (CountDownLatch countDownLatch,String location,WeatherServiceApi weatherServiceApi,
RedisTemplate redisTemplate,Integer redisWeather,String startDate,String endDate,Boolean onTheDay){
this.countDownLatch = countDownLatch;
this.location = location;
this.weatherServiceApi = weatherServiceApi;
this.redisTemplate = redisTemplate;
this.redisWeather = redisWeather;
this.startDate = startDate;
this.endDate = endDate;
this.onTheDay = onTheDay;
}
@Override
public List call() throws Exception {
try{
Long st = System.currentTimeMillis();
List forecastAirQuality = null;
if (onTheDay){
startDate = LocalDate.now().toString();
forecastAirQuality = queryAirQualityForecast(location,startDate,startDate);
}else{
forecastAirQuality = queryAirQualityForecast(location,startDate,startDate);
}
Long ed = System.currentTimeMillis();
log.debug("");
log.debug("==> AirQualityTask consume time :{},countDownLatch:{}",ed-st,countDownLatch);
return forecastAirQuality;
}catch (Exception e){
throw e;
}finally {
countDownLatch.countDown();
}
}
/**
* ( 3-7 )
*
* @param location
* @param startDate
* @param endDate
*/
public List queryAirQualityForecast(String location,String startDate, String endDate) {
LocalDate queryStart = DateUtil.stringTransLocalDate(startDate);
LocalDate queryEnd = DateUtil.stringTransLocalDate(endDate);
return queryForecastAir(location, queryStart, queryEnd);
}
private List queryForecastAir(String location, LocalDate queryStart, LocalDate queryEnd) {
Long period = DateUtil.localDatePeriod(queryStart, queryEnd);
if (period > 9) {
throw new BusinessException(ExceptionEnum.QUERY_OUT_RANGE);
}
boolean noCache = false;
String redisKey = null;
LocalDate startDate = null;
List list = new ArrayList<>();
for (int index = 0; index < period; index++) {
startDate = queryStart;
startDate = startDate.plusDays(index);
redisKey = RedisEnum.CpspWeather.LOCATION_DATE_SAVE_FORECAST_AIR_QUALITY.getKey(location, startDate.toString());
AirQuality airQuality = (AirQuality) redisTemplate.opsForValue().get(redisKey);
if (airQuality == null) {
list = new ArrayList<>();
noCache = true;
break;
} else {
list.add(airQuality);
}
}
if (noCache) {
List aqList = getAirQualityForecast(location, ServiceConstant.AIR_QUALITY_FORECAST);
if (CollectionUtils.isEmpty(aqList)) {
log.info("==> queryCacheDailyAir no data back!");
throw ExceptionEnum.CPSP_AIR_PARSE_ERROR.getException();
}
for (AirQuality aq : aqList) {
for (int day = 0; day < period; day++) {
startDate = queryStart;
startDate = startDate.plusDays(day);
if (startDate.toString().equals(aq.getQualityDate())) {
list.add(aq);
}
}
//setRedisForecastAir(aq,location);
redisKey = RedisEnum.CpspWeather.LOCATION_DATE_SAVE_FORECAST_AIR_QUALITY.getKey(location, aq.getQualityDate());
RedisOpsForValueAsync redisOpsForValueAsync = new RedisOpsForValueAsync(redisTemplate,redisWeather,redisKey,aq);
WeatherExecutorService executorService = new WeatherExecutorService();
executorService.submit(redisOpsForValueAsync);
}
}
return list;
}
, spring
// @Async
// public void setRedisForecastAir(AirQuality aq,String location){
// String redisKey = RedisEnum.CpspWeather.LOCATION_DATE_SAVE_FORECAST_AIR_QUALITY.getKey(location, aq.getQualityDate());
// redisTemplate.opsForValue().set(redisKey, aq, redisWeather, TimeUnit.SECONDS);
// }
/**
*
*
* @param location
* @param queryType
* @return
*/
private List getAirQualityForecast(String location, String queryType) {
JSONObject response = weatherServiceApi.queryAirQualityForecast(location);
//log.debug("==> cpsp getAirQualityForecast response : {}", response);
if (Objects.isNull(response)) {
log.error("==> cpsp response is null!");
throw new BusinessException(ExceptionEnum.CPSP_RESPONSE_TIMEOUT);
}
return CommonWeather.setAirQualityForecast(response,location);
}
}