Living in the concurrent world No.1

1. 重构场景1: 导入大数据量的文本文件(记录数>10000).
    性能分析: 系统要求能够接受xml, csv两类文件, 并且最后的处理统一按照xml来进行解析。由于系统架构的缘故, 目前是通过使用xslt将csv转换成xml文件。基本的过程是通过JAXB将xml结构解析成相应的对象, 然后调用统一的repository/domain操作, 将其插入数据库中. 从csv转换得到的xml文件需要从内存写入磁盘以便日后查询文件得到导入明细数据. 之前的结构在导入大数据量的csv文件时发现明显停滞. 经分析方法调用时间发现主要由于同一批次处理的csv文件记录过多(1000条), 而且由于生成的xml文件过大需要消耗更多的堆空间来处理对象的转换。
    改进方案: 减小批次的记录(100~300), 对每个批次的处理使用单独的线程而不是顺序处理批次.

 

2. 示例代码

    MessageReceiver类负责创建每个批次的具体执行线程。

 

class MessageReceiver implements Callable<Boolean> {
    
    /** The messages. */
    private List<Document> messages;
    
    /** The processed file. */
    private String file;
    
    /** The parms. */
    private Properties parms;
    
    /**
     * Instantiates a new message receiver.
     * 
     * @param messages the messages
     * @param file the processed file
     * @param parms the parms
     */
    public MessageReceiver (List<Document> messages, String file, Properties parms) {
        this.messages = messages;
        this.file = file;
        this.parms = parms;
    }
    
    /**
     * Call.
     * 
     * @return the boolean
     * 
     * @throws Exception the exception
     */
    public Boolean call() throws Exception
    {
        try {
            return addMessage();
        } catch (Exception e) {
            throw e;
        }
    }
    
    /**
     * Adds the message.
     * 
     * @return true, if successful
     */
    private boolean addMessage () {
        messages.add(TransformDocument(file, parms));
        return true;
    }
}

 

    receiveCSV 方法负责分解各个批次以及最后将各线程处理结果汇总计算。(省略关键业务逻辑, 只保留逻辑主干)

 

private List<Document> receiveCSV() {
    // omit the subsidiary process
    try
    {
        br = new BufferedReader(new FileReader(file));
        String text = "";
        while ((text = br.readLine()) != null)
        {
            fileContents.append(text + "\n");
            recordsNum++;
            if ((recordsNum % csvManageRecordNum) == 0)
            {
                String processedFile = getFileContent(fileContents);
                
                Future<Boolean> result = threads.submit(new MessageReceiver(messages, file, parms));
                results.add(result);
                
                fileContents = new StringBuilder();
            }
        }
    }
    catch (FileNotFoundException e)
    {
        //
    }
    catch (IOException e)
    {
        //
    }
    finally
    {
        // omit the dispose process
    }
    
    if ((recordsNum % csvManageRecordNum) != 0)
    {
        String file = getFileContent(fileContents);
        Future<Boolean> result = threads.submit(new MessageReceiver(messages, file, parms));
        results.add(result);
    }

    // Using thread pool to execute simultaneously
    try {
        for (Future<Boolean> result : results) {
            try {
                result.get();
            } catch (ExecutionException ee) {
                throw new Exception(ee);
            }
        }
    } catch (InterruptedException ie) {
        throw new Exception(ie);
    } finally {
        threads.shutdown();
    }
    
    return messages;
}