springboot的异步调用

Posted 2021-02-13 不死码农

package com.handsight.platform.fras.aapp;

import java.util.Locale;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.i18n.LocaleContextHolder;
import org.springframework.stereotype.Component;

import com.handsight.platform.common.util.LogUtil;
import com.handsight.platform.fras.data.StaticObject;
import com.handsight.platform.fras.thread.service.AsyncService;

@Component
public class StartupListener implements ApplicationListener<ContextRefreshedEvent> {

    private final static Logger logger = LoggerFactory.getLogger(StartupListener.class);

    @Autowired
    private AsyncService asyncService;

    public StartupListener() {
    };

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {

        // 地域语言
        setSystemDefaultLanguage();

        logger.info("---------------- Start execute Async thread completed.");
        // 删除用户的相关访问信息线程
        asyncService.execDeleteLogOutUserCacheInfo();

        // 启动图片批量发送线程
        asyncService.sendImageBatch();

        logger.info("---------------- End execute Async thread completed.");

    }

    /**
     * 设置系统默认语言
     */
    private  void setSystemDefaultLanguage() {
        Locale locale= LocaleContextHolder.getLocale();
        locale = Locale.CHINA;
//        if(!Constants.SYSTEM_DEFAULT_LANGUAGE.equals(locale.getLanguage()) ) {
//            locale = Locale.US;
//        }
        LocaleContextHolder.setLocale(locale);
        StaticObject.locale =locale;
        LogUtil.info("This language is {0}", StaticObject.locale.getLanguage());
    }

}

使用：

package com.handsight.platform.fras.thread.service;

import java.net.Socket;
import java.util.List;

import com.handsight.platform.common.exception.AppException;
import com.handsight.platform.fras.mgt.pojo.T_user;
import com.handsight.platform.fras.mgt.pojo.TransferAlgorithmReq;

/**
 * 异步线程的接口
 *
 * @author wangh
 *
 */
public interface AsyncService {


    /**
     * 批量发送图片
     *
     * @param userToken
     * @param facePhotoString
     * @throws AppException
     * @throws Exception
     */
    public void sendImageBatch();

    /**
     * 异步任务实现类: 用户退出后删除其可变用户令牌与账号的缓存信息；以及session信息
     */
    void execDeleteLogOutUserCacheInfo();

    /**
     * 异步任务实现类: 将硬件状态数据存入数据库
     */
    void execHardwareStatusDataToDBAsync();

    /**
     * 异步任务实现类:向算法发送任务 by http
     */
    public void workSendTaskThread(String userToken, String facePhotoString) throws AppException, Exception;

    /**
     * 异步任务实现类:向算法发送任务 by http
     *
     * @throws Exception
     */
    public void sendImageTaskThread(List<TransferAlgorithmReq> transferAlgorithmBeanList) throws AppException, Exception;

    /**
     * 异步任务实现类:保存用户令牌及 用户相关信息
     *
     * @param user
     * @throws Exception
     */
    public void saveUserInfo(T_user user) throws Exception;

    /**
     * 存储用户登录地点信息
     *
     * @param user
     * @return
     */
    public void saveLoginLocationInfo(T_user user);

    /**
     * 更新用户信息
     *
     * @param user
     * @return
     */
    public void updateUserInfo(T_user user) throws Exception;

    /***
     * 存储用户人脸信息及特征值
     *
     * @param user
     * @param userToken
     * @param currentFaceCode
     * @throws Exception
     */
    public void saveUserFaceCode(T_user user, String userToken, String currentFaceCode) throws Exception;

    /**
     * 异步任务实现类:接受算法产生的图片特征码 by http
     */
    public void workReciveResultThread(Socket socket);

    /**
     * 异步任务实现类:更新用户人脸特征库
     */
    public void updateUserFaceCodeListThread();

    /**
     *
     * 消息推送
     *
     * @param platform
     * @param pushKey
     * @param content
     * @throws Exception
     */
    public void pushMsg(String platform, String pushKey, String content) throws Exception;

    /**
     * 将月度登录失败次数加一
     */
    public void addOneMonthFailedNum(String userAccount) throws Exception;

    /**
     * 异步任务实现类:向算法发送任务
     */
    @Deprecated
    public void workSendTaskThread_skt(Socket socket);

    /**
     * 异步任务实现类:接受算法产生的图片特征码 socket
     */
    @Deprecated
    public void workReciveResultThread_skt(Socket socket);
}

实现：

package com.handsight.platform.fras.thread.service.impl;

import com.handsight.platform.common.exception.AppException;
import com.handsight.platform.common.util.HttpRequestUtil;
import com.handsight.platform.common.util.JsonUtil;
import com.handsight.platform.common.util.LogUtil;
import com.handsight.platform.common.util.UuidUtil;
import com.handsight.platform.fras.cache.UserCache;
import com.handsight.platform.fras.constant.Constants;
import com.handsight.platform.fras.constant.ErrorCodeMsg;
import com.handsight.platform.fras.data.StaticObject;
import com.handsight.platform.fras.mapper.UserMapper;
import com.handsight.platform.fras.mgt.pojo.T_user;
import com.handsight.platform.fras.mgt.pojo.TransferAlgorithmReq;
import com.handsight.platform.fras.mgt.pojo.TransferAlgorithmRes;
import com.handsight.platform.fras.mgt.pojo.UserFaceBean;
import com.handsight.platform.fras.pojo.MessageBean;
import com.handsight.platform.fras.service.CommonService;
import com.handsight.platform.fras.service.PushService;
import com.handsight.platform.fras.service.RedisService;
import com.handsight.platform.fras.thread.service.AsyncService;
import com.handsight.platform.fras.util.BeanUtil;
import com.handsight.platform.fras.util.EnumUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.NameValuePair;
import org.apache.http.message.BasicNameValuePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.handsight.platform.common.constant.Constants.HTTP_RES_CODE;
import static com.handsight.platform.common.constant.Constants.HTTP_RES_CONTENT;
import static com.handsight.platform.fras.constant.Constants.QUEUE_KEY_LOGOUT_USER_TOKEN;

@Service
public class AsyncServiceImpl implements AsyncService {

    private final static Logger logger = LoggerFactory.getLogger(AsyncServiceImpl.class);

    /** 定义一个每次要数据的大小(35K) */
    public final static int PART_SIZE = 35 * 1024;

    static boolean sendTaskThreadStart = false;

    @Autowired
    private RedisService redisService;

    @Autowired
    private PushService pushService;

    @Autowired
    private CommonService commonService;

    @Autowired
    private UserMapper userMapper;

    @Autowired
    private UserCache userCache;

    @Value("${fras.server.ip}")
    private String serverIp;

    @Value("${fras.send.work.port}")
    private String sendWorkPort;

    @Value("${fras.send.batch.list.size}")
    private int batchListSize;
    @Value("${fras.send.batch.interval.ms.times}")
    private int sendIntervalTimes;


    @Override
    @Async // ("frasAsyncServiceExecutor")
    public  void sendImageBatch() {
        long start = 0L;
         List<TransferAlgorithmReq> transferAlgorithmBeanList = null;
        while (true) {
            try {

                int cnt =0;
                // 每指定个数发送一次
                while (!StaticObject.imageQueue.isEmpty() ) {

                    if (StaticObject.transferAlgorithmBeanList.size() < batchListSize) {
                        redisService.hmSet("a---batchListSize", UuidUtil.uuid(), StaticObject.transferAlgorithmBeanList.size() ); // TODO
                        StaticObject.transferAlgorithmBeanList.add(StaticObject.imageQueue.take());
                        redisService.hmSet("cnt", UuidUtil.uuid(), ++cnt);
                        start = System.currentTimeMillis();
                    } else {
                        System.out.println("输出集合"+StaticObject.transferAlgorithmBeanList.size());
                        transferAlgorithmBeanList = BeanUtil.deepCopy(StaticObject.transferAlgorithmBeanList);
                        dealWithTask(transferAlgorithmBeanList);// 处理请求
                    }
                }
                // 不足指定个数每指定秒数发送一次
                if (StaticObject.transferAlgorithmBeanList.size() > 0 && (((System.currentTimeMillis() - start)) >= sendIntervalTimes)) {

                    transferAlgorithmBeanList = BeanUtil.deepCopy(StaticObject.transferAlgorithmBeanList);
                    dealWithTask(transferAlgorithmBeanList);// 处理请求
                } else {
                    Thread.sleep(10);
                }
            } catch (Exception e) {
                logger.error("提取特征码异常!", e);
            } finally {
            }
        }

    }

    /**
     * 处理任务队列
     * 当已发送的匹数大于3个时将延迟发送1秒
     *
     * @param transferAlgorithmBeanList
     */
    private void dealWithTask(List<TransferAlgorithmReq> transferAlgorithmBeanList) {
        try {
            redisService.hmSet("a---executer", UuidUtil.uuid(), transferAlgorithmBeanList.size());
            long num = redisService.increment(Constants.CACHE_KEY_ALGORITHM_TASK_REQ_NUM, 1L);
            if( num > 3L) {
                Thread.sleep(1000);
            }
            // 批量发送图片
            sendImageBatchTask(transferAlgorithmBeanList);

//            new Thread(new DealQueueThread(transferAlgorithmBeanList)).start();
            StaticObject.transferAlgorithmBeanList.clear();
        } catch (Exception e) {
            logger.error("特征码提取处理异常", e);
        } finally {
        }
    }

    /**
     * 批量发送图片
     *
     * @param transferAlgorithmBeanList
     * @throws AppException
     * @throws Exception
     */
    private void sendImageBatchTask(List<TransferAlgorithmReq> transferAlgorithmBeanList) throws AppException, Exception {

        long start = System.currentTimeMillis();
        String dataRequest = "";
        try {

            // 以表单形式向所有算法服务器发送指令 a
            dataRequest = JsonUtil.getJsonString(transferAlgorithmBeanList);
            List<NameValuePair> params = new ArrayList<NameValuePair>();
            NameValuePair pair = new BasicNameValuePair("data", dataRequest);
            params.add(pair);
            String httpUrlServer = Constants.HTTP_PREFIX + serverIp + Constants.COLON_SIGN + sendWorkPort + "/" + Constants.SEND_WORK_BATCH_METHOD + "/";

            logger.info("传输数据完毕, 耗时：" + ((System.currentTimeMillis() - start) ) + "ms,  num:" + transferAlgorithmBeanList.size() + "  size: " + dataRequest.length() / 1024);

            // 处理结果
            dealResponseResult(HttpRequestUtil.sendPostWithParamsForString(httpUrlServer, "", 1, params));

            logger.info("获取结果耗时：" + (System.currentTimeMillis() - start));
        } catch (AppException ape) {

            ErrorCodeMsg errorCodeMsg = EnumUtil.getByCode(ape.getCode(), ErrorCodeMsg.class);
            setExceptionForUser(errorCodeMsg, transferAlgorithmBeanList);
        }  catch (Exception e) {
            setExceptionForUser(ErrorCodeMsg.AI_SERVER_ABNORMAL, transferAlgorithmBeanList);
        } finally {
            redisService.increment(Constants.CACHE_KEY_ALGORITHM_TASK_REQ_NUM, -1L);
        }
    }

    /**
     * 处理http的响应结果
     *
     * @param resultMap
     */
    private void dealResponseResult(Map<String, Object> resultMap) throws Exception {

        if (resultMap.containsKey(Constants.EXCEPTION_ALGORITHM)) {

            String msg = commonService.getMessage(ErrorCodeMsg.AI_SERVER_ABNORMAL.getMsg());
            throw new AppException(ErrorCodeMsg.AI_SERVER_ABNORMAL.getCode(), msg);
        } else {

            String code = (String) resultMap.get(HTTP_RES_CODE);
            String content = (String) resultMap.get(HTTP_RES_CONTENT);

            if (!Constants.HTTP_STATUS_CODE_200.equals(code)) {
                logger.error("提取图片的特征码出现异常：code： " + code + " msg: " + content);
                String msg = commonService.getMessage(ErrorCodeMsg.LOGIN_FAILED_TRY_AGAIN.getMsg());
                throw new AppException(ErrorCodeMsg.LOGIN_FAILED_TRY_AGAIN.getCode(), msg);
            } else {

                List<TransferAlgorithmRes> resultList = JsonUtil.readJson2Array(content, TransferAlgorithmRes.class);
                for (TransferAlgorithmRes res : resultList) {
                    redisService.setForTimeMIN(Constants.FRAS_CACHE_KEY_ALGORITHM_RESULT + res.getId(), res, 10);
                }
            }
        }
    }

    /**
     *
     * 为每个用户设置异常时的信息
     *
     * @param erroCodeMsg
     * @param transferAlgorithmBeanList
     */
    private void setExceptionForUser(ErrorCodeMsg erroCodeMsg, List<TransferAlgorithmReq> transferAlgorithmBeanList) {

        int msgCode = erroCodeMsg.getCode();
        String msg = commonService.getMessage(erroCodeMsg.getMsg());
        for (TransferAlgorithmReq req : transferAlgorithmBeanList) {
            redisService.setForTimeMIN(Constants.CACHE_KEY_EXCEPTION_ALGORITHM + req.getId(), new MessageBean(msgCode, msg), 10);
        }
    }

    /**
     * 异步任务实现类:向算法发送任务 by http
     *
     * @throws Exception
     */
    @Override
//    @Async
    public void sendImageTaskThread(List<TransferAlgorithmReq> transferAlgorithmBeanList) throws AppException, Exception {

        long start = System.currentTimeMillis();
        String dataRequest = "";
        try {
            for (TransferAlgorithmReq req : transferAlgorithmBeanList) {
                System.out.println("http-------------------------------" + req.getId());
            }

            List<TransferAlgorithmRes> resultList = new ArrayList<TransferAlgorithmRes>();
            dataRequest = JsonUtil.getJsonString(transferAlgorithmBeanList);
            // 以表单形式向所有算法服务器发送指令
            List<NameValuePair> params = new ArrayList<NameValuePair>();
//            NameValuePair pair = new BasicNameValuePair("id", "1");
//            params.add(pair);
            NameValuePair pair = new BasicNameValuePair("data", dataRequest);
            params.add(pair);
            String httpUrlServer = Constants.HTTP_PREFIX + serverIp + Constants.COLON_SIGN + sendWorkPort + "/" + Constants.SEND_WORK_BATCH_METHOD + "/";
            logger.info("传输数据完毕, 2222耗时：" + ((System.currentTimeMillis() - start) / 1000) + "  num:" + transferAlgorithmBeanList.size() + "  size: " + dataRequest.length() / 1024);
            Map<String, Object> resultMap = HttpRequestUtil.sendPostWithParamsForString(httpUrlServer, "", 2, params);

            if (resultMap.containsKey(Constants.EXCEPTION_ALGORITHM)) {
                String msg = commonService.getMessage(ErrorCodeMsg.AI_SERVER_ABNORMAL.getMsg());
                throw new AppException(ErrorCodeMsg.AI_SERVER_ABNORMAL.getCode(), msg);
            } else {
                String code = (String) resultMap.get(HTTP_RES_CODE);
                String content = (String) resultMap.get(HTTP_RES_CONTENT);

                if (!Constants.HTTP_STATUS_CODE_200.equals(code)) {
                    logger.error("提取图片的特征码出现异常：code： " + code + " msg: " + content);
                    String msg = commonService.getMessage(ErrorCodeMsg.LOGIN_FAILED_TRY_AGAIN.getMsg());
                    throw new AppException(ErrorCodeMsg.LOGIN_FAILED_TRY_AGAIN.getCode(), msg);
                } else {
                    resultList = JsonUtil.readJson2Array(content, TransferAlgorithmRes.class);
                    for (TransferAlgorithmRes res : resultList) {
                        redisService.setForTimeMIN(Constants.FRAS_CACHE_KEY_ALGORITHM_RESULT + res.getId(), res, 10);
                    }
                }
            }
            logger.info("获取结果333耗时：" + (System.currentTimeMillis() - start) / 1000);
        } catch (Exception e) {
            throw e;
        } finally {
        }
    }

    /**
     * 异步任务实现类:向算法发送任务 by http
     *
     * @throws Exception
     */
    @Override
//    @Async("asyncServiceExecutor")、
    @Async
    public void workSendTaskThread(String userToken, String facePhotoString) throws AppException, Exception {

        Map<String, Object> resultMap = null;

        long start = 0;
        try {

            if (!sendTaskThreadStart) {
                sendTaskThreadStart = true;
                start = System.currentTimeMillis();
                while (true) {
                    if ((System.currentTimeMillis() - start) / 1000 > 5) {
                        break;
                    } else {
//                        TransferAlgorithmReq transferAlgorithmBean = (TransferAlgorithmReq) redisService.rightPop(Constants.QUEUE_TASK);
                        TransferAlgorithmReq transferAlgorithmBean = StaticObject.imageQueue.poll();
                        if (StaticObject.transferAlgorithmBeanList != null && StaticObject.transferAlgorithmBeanList.size() <= 4) {
                            if (transferAlgorithmBean != null) {
                                StaticObject.transferAlgorithmBeanList.add(transferAlgorithmBean);
                                logger.info("bean：" + transferAlgorithmBean.getId());
                            }
                        } else {
                            redisService.leftPush(Constants.QUEUE_TASK, transferAlgorithmBean);
                            break;
                        }
                    }
                }
            } else {
                return;
            }

            List<TransferAlgorithmRes> resultList = new ArrayList<TransferAlgorithmRes>();

            // 以表单形式向所有算法服务器发送指令
            List<NameValuePair> params = new ArrayList<NameValuePair>();
            NameValuePair pair = new BasicNameValuePair("id", userToken);
            params.add(pair);
            pair = new BasicNameValuePair("img", facePhotoString); // JsonUtil.getJsonString(StaticObject.transferAlgorithmBeanList) TODO
            params.add(pair);
            String httpUrlServer = Constants.HTTP_PREFIX + serverIp + Constants.COLON_SIGN + sendWorkPort + "/" + Constants.SEND_WORK_METHOD + "/";
            resultMap = HttpRequestUtil.sendPostWithParamsForString(httpUrlServer, "", 2, params);

            if (resultMap.containsKey(Constants.EXCEPTION_ALGORITHM)) {
                String msg = commonService.getMessage(ErrorCodeMsg.AI_SERVER_ABNORMAL.getMsg());
                throw new AppException(ErrorCodeMsg.AI_SERVER_ABNORMAL.getCode(), msg);
            } else {
                String code = (String) resultMap.get(HTTP_RES_CODE);
                String content = (String) resultMap.get(HTTP_RES_CONTENT);

                TransferAlgorithmRes sd = new TransferAlgorithmRes(userToken, 1002, "001,191,101"); // TODO
                TransferAlgorithmRes bean = JsonUtil.json2Obj(JsonUtil.getJsonString(sd), TransferAlgorithmRes.class); // TODO
                List<TransferAlgorithmRes> lst2 = new ArrayList<TransferAlgorithmRes>(); // TODO
                lst2.add(bean);// TODO
                content = JsonUtil.getJsonString(lst2);// TODO

                if (!Constants.HTTP_STATUS_CODE_200.equals(code)) {
                    logger.error("提取图片的特征码出现异常：code： " + code + " msg: " + content);
                    String msg = commonService.getMessage(ErrorCodeMsg.LOGIN_FAILED_TRY_AGAIN.getMsg());
                    throw new AppException(ErrorCodeMsg.LOGIN_FAILED_TRY_AGAIN.getCode(), msg);
                } else {
                    resultList = JsonUtil.readJson2Array(content, TransferAlgorithmRes.class);
                    for (TransferAlgorithmRes res : resultList) {
                        redisService.setForTimeMIN(Constants.FRAS_CACHE_KEY_ALGORITHM_RESULT + res.getId(), res, 10);
                    }

//                    redisService.setForTimeMIN(Constants.FRAS_CACHE_KEY_ALGORITHM_RESULT + userToken, content, 10);
                }
            }
            logger.info("传输数据完毕,耗时：" + (System.currentTimeMillis() - start + "  size:" + facePhotoString.length() / 1024));
        } catch (Exception e) {
            throw e;
        } finally {
            StaticObject.transferAlgorithmBeanList.clear();
            sendTaskThreadStart = false;
        }

//        return resultMap;
    }

    /**
     * 异步任务实现类:保存用户令牌及 用户相关信息
     *
     * @param user
     * @throws Exception
     */
    @Override
    @Async
    @Transactional(rollbackFor = Exception.class)
    public void saveUserInfo(T_user user) throws Exception {

        int cnt = 0;
        while (true) {
            try {
                try {
                    cnt++;
                    // 保存用户信息
                    commonService.saveUserInfo(user);
                    break;
                } catch (Exception e) {
                    if (cnt > 20) {
                        logger.error("数据保存失败，开始重试次数：" + cnt, e);
                        throw e;
                    } else {
                        Thread.sleep(5000);
                    }
                    logger.error("数据保存失败，开始重试次数：" + cnt, e);
                }
            } catch (Exception e) {
                logger.error("数据库连接发生异常,未保存数据 " + user.toString(), e);
                break;
            } finally {
            }
        }
    }

    /**
     * 存储用户登录地点信息
     *
     * @param user
     * @return
     */
    @Override
    @Async
    public void saveLoginLocationInfo(T_user user) {
        int cnt = 0;
        while (true) {
            try {
                try {
                    cnt++;
                    commonService.checkSQLReturnCode(userMapper.saveLoginLocationInfo(user));
                    break;
                } catch (Exception e) {
                    if (cnt > 20) {
                        logger.error("数据保存失败，开始重试次数：" + cnt, e);
                        throw e;
                    } else {
                        Thread.sleep(5000);
                    }
                    logger.error("数据保存失败，开始重试次数：" + cnt, e);
                }
            } catch (Exception e) {
                logger.error("数据库连接发生异常,未保存数据 " + user.toString(), e);
                break;
            } finally {
            }
        }
    }

    /***
     * 存储用户人脸信息及特征值
     *
     * @param user
     * @param userToken
     * @param currentFaceCode
     * @throws Exception
     */
    @Override
    public void saveUserFaceCode(T_user user, String userAccount, String currentFaceCode) throws Exception {

        int cnt = 0;
        while (true) {
            try {
                try {
                    cnt++;
                    commonService.saveUserFaceCode(user, userAccount, currentFaceCode);
                    break;
                } catch (Exception e) {
                    if (cnt > 20) {
                        logger.error("数据保存失败，开始重试次数：" + cnt, e);
                        throw e;
                    } else {
                        Thread.sleep(5000);
                    }
                    logger.error("数据保存失败，开始重试次数：" + cnt, e);
                }
            } catch (Exception e) {
                logger.error("数据库连接发生异常,未保存数据 " + user.toString(), e);
                break;
            } finally {
            }
        }
    }

    /**
     * 更新用户信息
     *
     * @param user
     * @return
     */
    @Override
    public void updateUserInfo(T_user user) throws Exception {

        int cnt = 0;
        while (true) {
            try {
                try {
                    cnt++;
                    userMapper.updateUserInfo(user);
                    // 更新用户信息
                    userCache.getUserInfo(user, true);
                    break;
                } catch (Exception e) {
                    if (cnt > 20) {
                        logger.error("数据保存失败，开始重试次数：" + cnt, e);
                        throw e;
                    } else {
                        Thread.sleep(5000);
                    }
                    logger.error("数据保存失败，开始重试次数：" + cnt, e);
                }
            } catch (Exception e) {
                logger.error("数据库连接发生异常,未保存数据 " + user.toString(), e);
                break;
            } finally {
            }
        }

    }

    /**
     * 异步任务实现类: 用户退出后删除其可变用户令牌与账号的缓存信息；以及session信息
     */
    @Override
    @Async
    public void execDeleteLogOutUserCacheInfo() {

        while (true) {
            String userTokenJson = null;
            // 加锁
            // lock.lock();
            try {
                try {
                    userTokenJson = (String) redisService.rightPop(QUEUE_KEY_LOGOUT_USER_TOKEN);
                    if (!waitForData(userTokenJson)) {
                        continue;
                    }
                } catch (Exception e) {
                }

                try {
                    // session
                    String sessionKey = Constants.SPRING_SESSION_NAME_NAMESPCE + userTokenJson;
                    if (redisService.hasKey(sessionKey)) {
                        redisService.delete(sessionKey);
                    }
                } catch (Exception e) {
                    logger.error("redis 连接异常", e);
                    redisService.leftPush(QUEUE_KEY_LOGOUT_USER_TOKEN, userTokenJson);
                    throw e;
                }
            } catch (Exception e) {
                logger.error("删除已退出的用户session信息失败，用户令牌：" + userTokenJson, e);
            } finally {
                // 解锁
                // lock.unlock();
            }
        }
    }

    /**
     * 异步任务实现类:接受算法产生的图片特征码 by http
     */
    @Override
//    @Async("asyncServiceExecutor")
    public void workReciveResultThread(Socket socket) {
        while (true) {
            try {

                /**
                 * 在从Socket的InputStream中接收数据时，像上面那样一点点的读就太复杂了， 有时候我们就会换成使用BufferedReader来一次读一行
                 *
                 * BufferedReader的readLine方法是一次读一行的，这个方法是阻塞的，直到它读到了一行数据为止程序才会继续往下执行，
                 * 那么readLine什么时候才会读到一行呢？直到程序遇到了换行符或者是对应流的结束符readLine方法才会认为读到了一行，
                 * 才会结束其阻塞，让程序继续往下执行。
                 * 所以我们在使用BufferedReader的readLine读取数据的时候一定要记得在对应的输出流里面一定要写入换行符（
                 * 流结束之后会自动标记为结束，readLine可以识别），写入换行符之后一定记得如果输出流不是马上关闭的情况下记得flush一下，
                 * 这样数据才会真正的从缓冲区里面写入。
                 */
                BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"));
                StringBuilder sb = new StringBuilder();
                String temp = "";
                while ((temp = br.readLine()) != null) {
                    sb.append(temp.substring(1, temp.length() - 1).replaceAll("‘", ""));
                    break;
                }

                String[] arr = sb.toString().split(Constants.COLON_SIGN);

                redisService.setForTimeMS(Constants.CACHE_KEY_RESULT + arr[0].trim(), arr[1].trim(), Constants.TIME_KEY_RESULT);

                logger.info("数据接受 --- Form Cliect[port:" + socket.getPort() + "] 消息内容:" + sb.toString());

            } catch (Exception e) {
                logger.error("work Recive Result Thread abnormal!", e);
            }
        }
    }

    /**
     * 异步任务实现类:向算法发送任务 by socket
     */
    @Deprecated
    @Override
//    @Async("asyncServiceExecutor")
    public void workSendTaskThread_skt(Socket socket) {

        String faceJson = "";
        while (true) {
            try {
                // 获取任务
                faceJson = (String) redisService.rightPop(Constants.QUEUE_TASK);
                if (StringUtils.isBlank(faceJson)) {
                    continue;
                }

//                UserFaceBean userFaceBean = JsonUtil.json2Obj(faceJson, UserFaceBean.class);

                DataOutputStream outputStream = null;
                outputStream = new DataOutputStream(socket.getOutputStream());
                byte[] jsonByte = faceJson.getBytes();

                logger.info("发送的数据长度为:" + jsonByte.length);
                Map<String, Integer> sizeMap = new HashMap<String, Integer>();
                int fileSize = jsonByte.length;
                sizeMap.put("size", fileSize);

                // 告诉服务器要发送文件的大小
                outputStream.write(JsonUtil.getJsonString(sizeMap).getBytes());

                int partCount = fileSize / PART_SIZE;
                int rest = fileSize % PART_SIZE;

                // 每次发送35K大小的数据
                for (int index = 0; index < partCount; index++) {
                    int beginIndex = index * PART_SIZE;
                    int endIndex = (index + 1) * PART_SIZE;
                    String temp = faceJson.substring(beginIndex, endIndex);
                    outputStream.write(temp.getBytes());
                }
                // 发送剩余的数据
                if (rest != 0) {
                    int beginIndex = partCount * PART_SIZE;
                    int endIndex = partCount * PART_SIZE + rest;
                    String temp = faceJson.substring(beginIndex, endIndex);
                    outputStream.write(temp.getBytes());
                }

                outputStream.flush();
                Thread.sleep(10);
                logger.info("传输数据完毕");
            } catch (Exception e) {
                redisService.leftPush(Constants.QUEUE_TASK, faceJson);
                logger.error("work Send Task Thread error!", e);
            }
        }
    }

    /**
     * 异步任务实现类:接受算法产生的图片特征码 by socket
     */
    @Deprecated
    @Override
//    @Async("asyncServiceExecutor")
    public void workReciveResultThread_skt(Socket socket) {
        while (true) {
            try {

                /**
                 * 在从Socket的InputStream中接收数据时，像上面那样一点点的读就太复杂了， 有时候我们就会换成使用BufferedReader来一次读一行
                 *
                 * BufferedReader的readLine方法是一次读一行的，这个方法是阻塞的，直到它读到了一行数据为止程序才会继续往下执行，
                 * 那么readLine什么时候才会读到一行呢？直到程序遇到了换行符或者是对应流的结束符readLine方法才会认为读到了一行，
                 * 才会结束其阻塞，让程序继续往下执行。
                 * 所以我们在使用BufferedReader的readLine读取数据的时候一定要记得在对应的输出流里面一定要写入换行符（
                 * 流结束之后会自动标记为结束，readLine可以识别），写入换行符之后一定记得如果输出流不是马上关闭的情况下记得flush一下，
                 * 这样数据才会真正的从缓冲区里面写入。
                 */
                BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"));
                StringBuilder sb = new StringBuilder();
                String temp = "";
                while ((temp = br.readLine()) != null) {
                    sb.append(temp.substring(1, temp.length() - 1).replaceAll("‘", ""));
                    break;
                }

                String[] arr = sb.toString().split(Constants.COLON_SIGN);

                redisService.setForTimeMS(Constants.CACHE_KEY_RESULT + arr[0].trim(), arr[1].trim(), Constants.TIME_KEY_RESULT);

                logger.info("数据接受 --- Form Cliect[port:" + socket.getPort() + "] 消息内容:" + sb.toString());

            } catch (Exception e) {
                logger.error("work Recive Result Thread abnormal!", e);
            }
        }
    }

    /**
     * 异步任务实现类: 将硬件状态数据存入数据库
     */
    @Override
//    @Async("asyncServiceExecutor")
    public void execHardwareStatusDataToDBAsync() {
        while (true) {
            String json = null;
            // 加锁
            // lock.lock();
            try {
                try {
//                    json = (String) redisService.rightPop(REDIS_HARDWARE_RUNNING_DATA);
                    if (!waitForData(json)) {
                        continue;
                    }
                } catch (Exception e) {
                }

                try {
//                    TODO
//                    hdwareStatusService.gatherHardwareStatusInfo(json);
                } catch (Exception e) {
                    if (!(e instanceof AppException)) {
                        LogUtil.error("", e);
//                        redisService.leftPush(REDIS_HARDWARE_RUNNING_DATA, json);
                    }
                    LogUtil.error("hardware status data insert db abnormal：" + json, e);
                }
            } catch (Exception e) {
                LogUtil.error(json, e);
            } finally {
                // 解锁
                // lock.unlock();
            }
        }
    }

    /**
     * 异步任务实现类:更新用户人脸特征库
     */
    @Override
//    @Async("asyncServiceExecutor")
    public void updateUserFaceCodeListThread() {

        while (true) {
            String userFaceJson = null;
            // 加锁
            // lock.lock();
            try {
                try {
                    userFaceJson = (String) redisService.rightPop(Constants.CACHE_KEY_UPDATE_FACE_CODE);
                    if (!waitForData(userFaceJson)) {
                        continue;
                    }
                } catch (Exception e) {
                    logger.error("redis exception", e);
                }

                try {
//                    TODO
//                    hdwareStatusService.gatherHardwareStatusInfo(json);
                    updateUserFaceCode(userFaceJson);
                } catch (Exception e) {
                    if (!(e instanceof AppException)) {
                        logger.error("", e);
                        redisService.leftPush(Constants.CACHE_KEY_UPDATE_FACE_CODE, userFaceJson);
                    }
                    LogUtil.error("hardware status data insert db abnormal：" + userFaceJson, e);
                }
            } catch (Exception e) {
                logger.error("system exception", e);
            } finally {
                // 解锁
                // lock.unlock();
            }
        }

    }

    /**
     * 消息推送
     *
     * @param platform
     * @param pushKey
     * @param content
     * @throws Exception
     * @return
     */
    @Override
//     @Async("asyncServiceExecutor")
    @Async
    public void pushMsg(String platform, String pushKey, String content) throws Exception {
        // 消息推送
        pushService.pushMsg(platform, pushKey, content);
    }

    /**
     * 将月度登录失败次数加一
     */
    @Override
//    @Async("asyncServiceExecutor")
    @Async
    public void addOneMonthFailedNum(String userAccount) throws Exception {
        // 更新月度登录失败次数
        commonService.checkSQLReturnCode(userMapper.addOneMonthFailedNum(userAccount));

        // 更新用户月度登录失败次数
        T_user param = new T_user();
        param.setUserAccount(userAccount);
        userCache.getUserInfo(param, true);
    }

    /**
     * 更新用户特征码
     *
     * @param userFaceJson
     * @throws Exception
     */
    private void updateUserFaceCode(String userFaceJson) throws Exception {
        UserFaceBean faceBean = JsonUtil.json2Obj(userFaceJson, UserFaceBean.class);
        commonService.saveUserFaceCode(null, faceBean.getUserToken(), faceBean.getFeatureCodes());
    }

    /**
     * 数据为空时, 返回false；否则，返回true
     *
     * @param json
     * @return
     */
    private boolean waitForData(String json) {
        if (json == null) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                LogUtil.error(json, e);
            }
            return false;
        } else {
            return true;
        }
    }
}