tối ưu mã nguồn saupublic class CallStatusThread implements Runnable private HashMapInteger ListCampaignBO campaignMap; private HashMapInteger String templateMap; private BotServiceConsumer

public class CallStatusThread implements Runnable {

private final HashMap<Integer, List<CampaignBO>> campaignMap;
private final HashMap<Integer, String> templateMap;
private final BotServiceConsumer consumer;
private static final Logger LOGGER = Logger.getLogger(CallStatusThread.class);
private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(Constants.UpdateCall.POOL_SIZE);
private final Thread thread;

public CallStatusThread() {
    thread = new Thread(this);

    campaignMap = new HashMap<>();
    templateMap = new HashMap<>();

    campaignMap = getBotCampaignMap();
    templateMap = getBotTemplateMap();

    consumer = new BotServiceConsumer(Constants.UpdateCall.Kafka.BROKER, Constants.UpdateCall.Kafka.STATUS_TOPIC, Constants.UpdateCall.Kafka.STATUS_GROUP, Constants.UpdateCall.Kafka.MAX_POLL_RECORDS);
}

private HashMap<Integer, List<CampaignBO>> getBotCampaignMap() {
    List<CampaignBO> campaignBOS = new CampaignDAL().getCampaignsConfig();
    HashMap<Integer, List<CampaignBO>> botCampaignMap = new HashMap<>();
    Set<Integer> botIdSet = new HashSet<>();

    for (CampaignBO campaignBO : campaignBOS) {
        botIdSet.add(campaignBO.getBotId());
    }

    for (Integer botId : botIdSet) {
        botCampaignMap.put(botId, new CampaignDAL().getCampaignsConfig(botId));
    }
    return botCampaignMap;
}

private HashMap<Integer, String> getBotTemplateMap() {
    List<CallOutMessageBO> callOutMessageBOS = new CallOutMessageDAL().getMessagesBO();
    HashMap<Integer, String> botTemplateMap = new HashMap<>();
    for (CallOutMessageBO callOutMessageBO : callOutMessageBOS) {
        int botId = callOutMessageBO.getBotId();
        if (botTemplateMap.get(botId) == null) {
            botTemplateMap.put(botId, callOutMessageBO.getTemplatePostCall());
        }
    }
    return botTemplateMap;
}

private void processingCallStatus() throws ParseException {
    consumer.subscribe();
    CallOutInfoDAL connector = new CallOutInfoDAL();
    LOGGER.info("Starting UpdateCallStatusThread ...");

    while (true) {
        ConsumerRecords<String, String> records = consumer.getRecords();
        if (records == null || records.isEmpty()) {
            continue;
        }
        LOGGER.info("Consume " + records.count() + " messages");
        ArrayList<CallOutInfoBO> callStatusBOS = new ArrayList<>();

        for (ConsumerRecord<String, String> record : records) {
            try {
                CallOutInfoBO bo = MappingObjectJson.convertJsonToObject(record.value(), CallOutInfoBO.class);
                callStatusBOS.add(bo);

                int botId = bo.getBotId();

                List<CampaignBO> campaignBOs = campaignMap.get(botId);
                int maxRetry = campaignBOs.get(0).getRetryTime();

                int callTimes = bo.getTimeCallOfDay();

                String template = templateMap.get(botId);
                if (template == null || template.isEmpty()) {
                    LOGGER.info("Don't Update with botId = " + botId + ", " + bo);
                } else {
                    EXECUTOR_SERVICE.execute(() -> {
                        try {
                            boolean isUpdated = update(bo, template, botId);
                            LOGGER.info("Update result : " + isUpdated + ", botId=" + botId + ", " + bo);
                        } catch (Exception e) {
                            LOGGER.error("Update ERROR, botId = " + botId + ", " + bo, e);
                        }
                    });
                }
            } catch (Exception e) {
                LOGGER.error("Process message " + record.value() + " error", e);
            }
        }

        CallOutInfoBO[] callStatusArray = new CallOutInfoBO[callStatusBOS.size()];
        callStatusBOS.toArray(callStatusArray);
        connector.insert(callStatusArray);
    }
}

public boolean update(CallOutInfoBO bo, String template, int botId) throws Exception {
    if (botId == 9) {
        return HttpClientFactory.updateCallStatusToOrder(bo);
    }
    return true;
}

public void start() {
    thread.start();
}

@Override
public void run() {
    try {
        processingCallStatus();
    } catch (ParseException e) {
        LOGGER.error("processingCallStatus error", e);
    }
}

}