資訊專欄INFORMATION COLUMN
摘要：說明這里用到的項(xiàng)目都是基于的項(xiàng)目。但同時(shí)，它和其他的分布式文件系統(tǒng)的區(qū)別也是很明顯的。能提供高吞吐量的數(shù)據(jù)訪問，非常適合大規(guī)模數(shù)據(jù)集上的應(yīng)用。放寬了一部分約束，來實(shí)現(xiàn)流式讀取文件系統(tǒng)數(shù)據(jù)的目的。是項(xiàng)目的一部分。
關(guān)鍵詞
Java、PHP、hdfs、mqrocket、excel、poi、報(bào)表
需求背景在業(yè)務(wù)需求方面，每個(gè)企業(yè)或多或少都會(huì)有報(bào)表導(dǎo)出的作業(yè)，量少則可是使用輸出流或者字符串的輸出即可完成，只要指定respose的相應(yīng)Content-Type即可。如果大量的數(shù)據(jù)需要導(dǎo)出，尤其是訂單這類業(yè)務(wù)邏輯復(fù)雜的報(bào)表，導(dǎo)出的時(shí)候需要加入各種條件和權(quán)限，從數(shù)據(jù)處理方面就已經(jīng)很費(fèi)力了，更何況導(dǎo)出的需求不是一天兩天，而是半月一月的數(shù)據(jù)量，小公司的業(yè)務(wù)，數(shù)量級(jí)也可能達(dá)到了十多萬。
function generateExcel($filename, $header, array &$data)
{
generateDownHeader($filename);
$rs = "| ".$v." | "; } $rs .= "|
|---|---|
| ".$v." | "; } else { $rs .= "".$v." | "; } } $rs .= "
這十多萬的數(shù)據(jù)，如果使用一般的方法（上面代碼所示）或許是不可行的（其他一般方法沒有嘗試過），php處理中一般使用curl調(diào)用接口，nginx服務(wù)器和php中的curl請求超時(shí)一般都是30s，30s處理1w條數(shù)據(jù)的導(dǎo)出工作，如果服務(wù)器的性能好，并且是多核的，可以使用multi_curl多線程處理，如果服務(wù)器的性能不是很好，這種處理方法或許更耗時(shí)。
下面是我使用的curl處理接口數(shù)據(jù)：
function curl($url, $option = null, $method = "POST", $getCode = false, $header = [])
{
$curl = curl_init ();
curl_setopt($curl, CURLOPT_URL, $url);
curl_setopt($curl, CURLOPT_TIMEOUT, 30);
if (!array_key_exists("Content-Type", $header)) {
$header["Content-Type"] = "application/json;charset=UTF-8";
}
$headers = [];
if ($header) {
foreach ($header as $k=>$v) {
$headers[] = $k.": ".$v;
}
}
curl_setopt($curl, CURLOPT_HTTPHEADER, $headers);
if ($option) {
if (is_array($option)) {
$option = json_encode($option);
}
curl_setopt($curl, CURLOPT_POSTFIELDS, $option);
}
curl_setopt($curl, CURLOPT_RETURNTRANSFER, 1);
curl_setopt($curl, CURLOPT_CUSTOMREQUEST, $method);
$result = curl_exec($curl);
if ($getCode) {
$curl_code = curl_getinfo($curl, CURLINFO_HTTP_CODE);
$message = self::isJson($result) ? json_decode($result, true) : $result;
$result = ["code" => $curl_code];
if (isset($message["exception"]) && count($message) == 1) {
$result["exception"] = $message["exception"];
$result["result"] = null;
} else {
$result["result"] = $message;
}
}
curl_close($curl);
return $result;
}
因?yàn)閿?shù)據(jù)量大，后來改為多線程：
function curlMulti(array $urls, $options = null, $method = "POST", $getCode = false, $header = [])
{
$mh = curl_multi_init();
// 添加curl批處理會(huì)話
$handles = $contents = [];
foreach ($urls as $key => $url) {
$handles[$key] = curl_init($url);
curl_setopt($handles[$key], CURLOPT_RETURNTRANSFER, 1);
curl_setopt($handles[$key], CURLOPT_TIMEOUT, 30);
curl_setopt($handles[$key], CURLOPT_CUSTOMREQUEST, $method);
if (!array_key_exists("Content-Type", $header)) {
$header["Content-Type"] = "application/json;charset=utf-8";
}
$headers = [];
if ($header) {
foreach ($header as $k => $val) {
$headers[] = $k.": ".$val;
}
}
curl_setopt($handles[$key], CURLOPT_HTTPHEADER, $headers);
if ($options) {
if (is_array($options)) {
$options = json_encode($options);
}
curl_setopt($handles[$key], CURLOPT_POSTFIELDS, $options);
}
curl_multi_add_handle($mh, $handles[$key]);
}
// 執(zhí)行批處理句柄
/*$active = null;
do{
$mrc = curl_multi_exec($mh, $active);
} while ($mrc == CURLM_CALL_MULTI_PERFORM);
while ($active and $mrc == CURLM_OK) {
if (curl_multi_select($mh) === -1) {
usleep(100);
do {
$mrc = curl_multi_exec($mh, $active);
}while($mrc == CURLM_CALL_MULTI_PERFORM);
}
}// 獲取批處理內(nèi)容
$errors = [];
foreach ($handles as $k => $ch) {
$errors[$k] = curl_error($ch);
$content = curl_multi_getcontent($ch);
if ($getCode) {
$content = curl_errno($ch) == 0 && self::isJson($content)? json_decode($content,true) : [];
}
$contents = array_merge($contents,$content);
}
$info = curl_multi_info_read($mh);*/
$output = $errors = $infos = [];
do {
while (($execrun = curl_multi_exec($mh, $running)) == CURLM_CALL_MULTI_PERFORM);
if ($execrun != CURLM_OK)
break;
while ($done = curl_multi_info_read($mh)) {
$info= curl_getinfo($done["handle"]);
$infos["http_code"][] = $info["http_code"];
$result["code"] = $info["http_code"];
$infos["url"][] = $info["url"];
$errors[] = curl_error($done["handle"]);
$output = self::isJson(curl_multi_getcontent($done["handle"])) ?
array_merge($output, json_decode(curl_multi_getcontent($done["handle"]),true)) : $output;
if ($running)
curl_multi_select($mh, 30);
}
} while ($running);
$result["result"] = $output;
$result["exception"] = $errors;
$result["info"] = $infos;
foreach ($handles as $ch) {
curl_multi_remove_handle($mh, $ch);
}
curl_multi_close($mh);
return $result;
}
上面的代碼中有一段代碼是注釋掉的，按照道理來說，上面的代碼執(zhí)行的結(jié)果應(yīng)該和下面的一樣，事實(shí)證明，卻是執(zhí)行的結(jié)果是一樣，我這里說的結(jié)果不是多線程返回的結(jié)果，既然是多線程，那么不同的線程競爭到資源也是不一樣的，返回結(jié)果出現(xiàn)了混亂，導(dǎo)出的excel數(shù)據(jù)并不是根據(jù)某種排序而排序的，也就是你不知道那個(gè)線程先返回了結(jié)果，這是問題一，其二，在導(dǎo)出的過程中，發(fā)現(xiàn)不同程度的丟失數(shù)據(jù)，加熱管每個(gè)線程500條數(shù)據(jù)，結(jié)果在驗(yàn)證數(shù)據(jù)時(shí)，發(fā)現(xiàn)僅僅返回了300多條數(shù)據(jù)，數(shù)據(jù)變動(dòng)不一致，第三，過多的數(shù)據(jù)，依然造成nginx服務(wù)器超時(shí)，錯(cuò)誤code 504。
PS: 為什么在php的中沒有使用phpexcel第三方包，原因很簡單，測試發(fā)現(xiàn)，phpexcel太耗內(nèi)存，機(jī)器吃不消，所以就沒用。
初步解決方案既然php的多線程方案不能解決問題，只能找其他的辦法，最可靠的也是大家都能想到的，就是隊(duì)列處理，把導(dǎo)出請求放入到隊(duì)列中，直接返回給客戶端，告訴客戶業(yè)務(wù)正在處理，然后具體的導(dǎo)出交由消費(fèi)端處理，最后把結(jié)果反饋到客戶端。
我們都知道php的隊(duì)列有很多，常用的比如Swoole，Workman以及Gearman等。我選擇了Gearman，因?yàn)榉奖?，而Swoole原來在我們的項(xiàng)目中，后來被踢掉了，不知原由。
Gearman服務(wù)端work的代碼demo：
addServer();
$worker->addFunction("export", function (GearmanJob $job) {
$workload = $job->workload();
if (($data = $this->parseJson($workload)) == false) {
return AppHelper::encodeJson(["code" => "-1", "result" => null, "exception" => "參數(shù)錯(cuò)誤"]);
}
$user = isset($data["user"]) && !empty($data["user"]) ? $data["user"] : "guest";
$path = dirname(Yii::$app->basePath) . "/backend/downloads/" . sha1($user) . "/" . date("Y-m-d") . "/";
$filename = isset($data["filename"]) && !empty($data["filename"]) ? $data["filename"] : date("Y-m-d") . "-order.xls";
$rs = $this->getData($data["type"]["data"], $data["type"]["count"], $data["api"], $data["params"]);
$this->writeExcel($path, $filename, $rs, $data["header"]);
return 200;
});
//無際循環(huán)運(yùn)行，gearman內(nèi)部已有處理，不會(huì)出現(xiàn)占用過高死掉的情況
while ($worker->work()) {
if ($worker->returnCode() !== GEARMAN_SUCCESS) {
echo "error" . PHP_EOL;
}
}
}
public function parseJson($str)
{
$data = json_decode($str, true);
return (json_last_error() == JSON_ERROR_NONE) ? $data : false;
}
public function writeExcel($path, $filename, $data, $header)
{
if ($this->mkDir($path)) {
$data = $this->assembleData($data);
$rs = $this->generateExcel($header, $data);
file_put_contents(rtrim($path, "/") . "/" . $filename, $rs);
} else {
echo "目錄不存在,寫文件錯(cuò)誤!";
}
return;
}
public function getData($dataApi, $countApi, $api, $params)
{
$start = microtime(true);
$count = AppHelper::getData($api . $countApi . "?" . http_build_query($params));
echo $api . $countApi . "?" . http_build_query($params).PHP_EOL;
echo "總條數(shù):" . $count . PHP_EOL;
$params["perpage"] = 500;
$times = ceil($count / $params["perpage"]);
$data = [];
if ($count > 0) {
for ($i = 0; $i < $times; $i++) {
$params["page"] = $i + 1;
$rs = AppHelper::getData($api . $dataApi . "?" . http_build_query($params));
$data = array_merge($data, $rs);
}
}
$end = microtime(true);
echo "花費(fèi)時(shí)間:" . ($end - $start) . PHP_EOL;
return $data;
}
public function generateExcel($header, array &$data)
{
$rs = "| " . $v . " | "; } $rs .= "|
|---|---|
| " . $v . " | "; } else { $rs .= "" . $v . " | "; } } $rs .= "
Gearman的Client端的代碼：
addServer("127.0.0.1", 4730);
$client->setCompleteCallback(completeCallBack);
$result2 = $client->doBackground("export", $str);//異步進(jìn)行，只返回處理句柄。
// $result1 = $client->do("export", "do");//do是同步進(jìn)行，進(jìn)行處理并返回處理結(jié)果。
// $result3 = $client->addTask("export", "addTask");//添加任務(wù)到隊(duì)列,同步進(jìn)行？通過添加task可以設(shè)置回調(diào)函數(shù)。
// $result4 = $client->addTaskBackground("export", "addTaskBackground");//添加后臺(tái)任務(wù)到隊(duì)列，異步進(jìn)行？
$client->runTasks();//運(yùn)行隊(duì)列中的任務(wù)，只是do系列不需要runTask()
return $result2;
}
//綁定回調(diào)函數(shù)，只對(duì)addTask有效
function completeCallBack($task)
{
echo "CompleteCallback！handle result:".$task->data()."
";
}
ps:要運(yùn)行上面的代碼，需要在服務(wù)器或者本地安裝Gearman服務(wù)，并且需要安裝php_gearman擴(kuò)展，安裝教程自行搜索。
如果你的業(yè)務(wù)邏輯不復(fù)雜，到此可以導(dǎo)出幾萬條數(shù)據(jù)綽綽有余了，然而，我的問題并沒有因此而解決，上司說，不想用Gearman隊(duì)列處理，最好還是java處理。嗯，沒關(guān)系，我喜歡這種在技術(shù)中跳來跳去的解決問題，既然不滿足上司的需求，那就另行方案。
MqRocket+HDFS+POI說明：這里用到的java項(xiàng)目都是基于spring+dubbo/dubbox的項(xiàng)目。所用到的配置或者注解均在spring的相關(guān)配置和注解范疇，除了mapper的配置和注解。
三個(gè)項(xiàng)目：
mq項(xiàng)目：提供rest服務(wù)，發(fā)送消息（@rxl）
biz項(xiàng)目：提供dubbo、restfull接口，處理業(yè)務(wù)（@lee）
data項(xiàng)目：處理數(shù)據(jù)導(dǎo)出
如上，三個(gè)項(xiàng)目分別是不同的工程師所寫，我們不關(guān)心怎么實(shí)現(xiàn)的，只需知道我們能使用每個(gè)功能即可。
mq提供的restfull接口：@Path("/message")
@Produces({ContentType.APPLICATION_JSON_UTF_8})
@Component("sendMessageService")
public class SendMessageImpl implements SendMessageService{
@Resource
public IProducer producer;
@PUT
@Path("send")
@Consumes({MediaType.APPLICATION_JSON})
@Override
public void sendMessage(Message message) {
System.out.println("message" + message.getMessage());
producer.send(message.getTopic(),message.getKey(),message.getMessage());
}
}
這樣我們在php后臺(tái)通過put方式，調(diào)用該接口，將需要處理的數(shù)據(jù)發(fā)送給導(dǎo)出處理服務(wù)端。發(fā)送put請求可以使用curl強(qiáng)大的request功能。
curl_setopt($curl, CURLOPT_CUSTOMREQUEST, "PUT");
假如mq提供的rest接口是：http://localhost:8018/mq/message/send,我們需要傳遞一個(gè)json字符串，該字符串原型是一個(gè)關(guān)聯(lián)數(shù)組，數(shù)組的key分別為“topic”、“key”和“message”，topic是消息的主題，需要指定的mq主題去消費(fèi)，key是消息的key，該topic下面會(huì)有很多key，因此，我們的消費(fèi)方即數(shù)據(jù)導(dǎo)出方需要根據(jù)key做判斷處理。message里面就是具體的一下參數(shù)，比如需要導(dǎo)出哪些字段，比如文件上傳服務(wù)器地址等等信息。
$message = [
"topic" => "order_export",
"key" => "order_tag_" . $orderNo,
"message" => [
"params" => [
...
],
"headers" => [
...
],
"options" => [
...
],
],
];
完整的接口請求：
http://localhost:8018/mq/message/send?{"topic":"order_export","key":"order_tag_","message":{"params":[],"header":[],"options":[]}}
poi工具類封裝Java的Excel API很多，唯獨(dú)Apache POI這款使用最方便最靈活（或許其他的沒有使用過）。
HSSF is the POI Project"s pure Java implementation of the Excel "97(-2007) file format. XSSF is the POI Project"s pure Java implementation of the Excel 2007 OOXML (.xlsx) file format.
HSSF and XSSF provides ways to read spreadsheets create, modify, read and write XLS spreadsheets. They provide:
low level structures for those with special needs
an eventmodel api for efficient read-only access
a full usermodel api for creating, reading and modifying XLS files
在gradle引入poi包：
// java excel api compile "org.apache.poi:poi:3.10.1" compile "org.apache.poi:poi-ooxml:3.9"
package cn.test.web.utils;
import cn.test.util.Utils;
import org.apache.commons.io.FilenameUtils;
import org.apache.poi.hssf.record.crypto.Biff8EncryptionKey;
import org.apache.poi.hssf.usermodel.HSSFFont;
import org.apache.poi.hssf.usermodel.HSSFFooter;
import org.apache.poi.hssf.usermodel.HSSFHeader;
import org.apache.poi.hssf.usermodel.HSSFWorkbook;
import org.apache.poi.openxml4j.exceptions.InvalidFormatException;
import org.apache.poi.poifs.filesystem.POIFSFileSystem;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.CellStyle;
import org.apache.poi.ss.usermodel.Font;
import org.apache.poi.ss.usermodel.Footer;
import org.apache.poi.ss.usermodel.Header;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.ss.usermodel.WorkbookFactory;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Properties;
/**
* Created with test-data
* User zhoujunwen
* Date 16/8/11
* Time 下午5:02
*/
public class POIUtils {
private static final short HEADER_FONT_SIZE = 16; // 大綱字體
private static final short FONT_HEIGHT_IN_POINTS = 14; // 行首字體
public static Workbook createWorkbook(String file) {
String ext = FilenameUtils.getExtension(CommonUtils.getFileName(file));
Workbook wb = null;
switch (ext) {
case "xls":
wb = createHSSFWorkbook();
break;
case "xlsx":
wb = createXSSFWorkbook();
break;
default:
wb = createHSSFWorkbook();
}
return wb;
}
public static Workbook createWorkbookByIS(String file, InputStream inputStream) {
String ext = FilenameUtils.getExtension(CommonUtils.getFileName(file));
Workbook wb = null;
try {
switch (ext) {
case "xls":
wb = new HSSFWorkbook(inputStream);
break;
case "xlsx":
wb = new XSSFWorkbook(inputStream);
break;
default:
wb = new HSSFWorkbook(inputStream);
}
} catch (IOException e) {
e.printStackTrace();
}
return wb;
}
public static Workbook writeFile(Workbook wb, String file) {
if (wb == null || Utils.isEmpty(file)) {
return null;
}
FileOutputStream out = null;
try {
out = new FileOutputStream(file);
wb.write(out);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (out != null) {
try {
out.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return wb;
}
public static Workbook createHSSFWorkbook() {
//生成Workbook
HSSFWorkbook wb = new HSSFWorkbook();
//添加Worksheet（不添加sheet時(shí)生成的xls文件打開時(shí)會(huì)報(bào)錯(cuò)）
@SuppressWarnings("unused")
Sheet sheet = wb.createSheet();
return wb;
}
public static Workbook createXSSFWorkbook() {
XSSFWorkbook wb = new XSSFWorkbook();
@SuppressWarnings("unused")
Sheet sheet = wb.createSheet();
return wb;
}
public static Workbook openWorkbook(String file) {
FileInputStream in = null;
Workbook wb = null;
try {
in = new FileInputStream(file);
wb = WorkbookFactory.create(in);
} catch (InvalidFormatException | IOException e) {
e.printStackTrace();
} finally {
try {
if (in != null) {
in.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
return wb;
}
public static Workbook openEncryptedWorkbook(String file, String password) {
FileInputStream input = null;
BufferedInputStream binput = null;
POIFSFileSystem poifs = null;
Workbook wb = null;
try {
input = new FileInputStream(file);
binput = new BufferedInputStream(input);
poifs = new POIFSFileSystem(binput);
Biff8EncryptionKey.setCurrentUserPassword(password);
String ext = FilenameUtils.getExtension(CommonUtils.getFileName(file));
switch (ext) {
case "xls":
wb = new HSSFWorkbook(poifs);
break;
case "xlsx":
wb = new XSSFWorkbook(input);
break;
default:
wb = new HSSFWorkbook(poifs);
}
} catch (IOException e) {
e.printStackTrace();
}
return wb;
}
/**
* 追加一個(gè)sheet,如果wb為空且isNew為true,創(chuàng)建一個(gè)wb
*
* @param wb
* @param isNew
* @param type 創(chuàng)建wb類型,isNew為true時(shí)有效 1:xls,2:xlsx
* @return
*/
public static Workbook appendSheet(Workbook wb, boolean isNew, int type) {
if (wb != null) {
Sheet sheet = wb.createSheet();
} else if (isNew) {
if (type == 1) {
wb = new HSSFWorkbook();
wb.createSheet();
} else {
wb = new XSSFWorkbook();
wb.createSheet();
}
}
return wb;
}
public static Workbook setSheetName(Workbook wb, int index, String sheetName) {
if (wb != null && wb.getSheetAt(index) != null) {
wb.setSheetName(index, sheetName);
}
return wb;
}
public static Workbook removeSheet(Workbook wb, int index) {
if (wb != null && wb.getSheetAt(index) != null) {
wb.removeSheetAt(index);
}
return wb;
}
public static Workbook insert(Workbook wb, String sheetName, int row, int start,
List columns) {
if (row == 0 || wb == null) return wb;
for (int i = start; i < (row + start); i++) {
Row rows = wb.getSheet(sheetName).createRow(i);
if (columns != null && columns.size() > 0) {
for (int j = 0; j < columns.size(); j++) {
Cell ceil = rows.createCell(j);
ceil.setCellValue(String.valueOf(columns.get(j)));
}
}
}
return wb;
}
/**
* 設(shè)置excel頭部
*
* @param wb
* @param sheetName
* @param columns 比如:["國家","活動(dòng)類型","年份"]
* @return
*/
public static Workbook setHeader(Workbook wb, String sheetName, List columns) {
if (wb == null) return null;
if (sheetName == null) {
sheetName = wb.getSheetAt(0).getSheetName();
}
return setHeaderStyle(insert(wb, sheetName, 1, 0, columns), sheetName);
}
/**
* 插入數(shù)據(jù)
*
* @param wb Workbook
* @param sheetName sheetName,默認(rèn)為第一個(gè)sheet
* @param start 開始行數(shù)
* @param data 數(shù)據(jù),List嵌套List ,比如:[["中國","奧運(yùn)會(huì)",2008],["倫敦","奧運(yùn)會(huì)",2012]]
* @return
*/
public static Workbook setData(Workbook wb, String sheetName, int start,
List data) {
if (wb == null) return null;
if (sheetName == null) {
sheetName = wb.getSheetAt(0).getSheetName();
}
if (data != null || data.size() > 0) {
if (data instanceof List) {
int s = start;
for (Object columns : data) {
insert(wb, sheetName, data.size() - (s - 1), s, (List) columns);
s++;
}
}
}
return wb;
}
/**
* 移除某一行
*
* @param wb
* @param sheetName sheet name
* @param row 行號(hào)
* @return
*/
public static Workbook delRow(Workbook wb, String sheetName, int row) {
if (wb == null) return null;
if (sheetName == null) {
sheetName = wb.getSheetAt(0).getSheetName();
}
Row r = wb.getSheet(sheetName).getRow(row);
wb.getSheet(sheetName).removeRow(r);
return wb;
}
/**
* 移動(dòng)行
*
* @param wb
* @param sheetName
* @param start 開始行
* @param end 結(jié)束行
* @param step 移動(dòng)到那一行后(前) ,負(fù)數(shù)表示向前移動(dòng)
* moveRow(wb,null,2,3,5); 把第2和3行移到第5行之后
* moveRow(wb,null,2,3,-1); 把第3行和第4行往上移動(dòng)1行
* @return
*/
public static Workbook moveRow(Workbook wb, String sheetName, int start, int end, int step) {
if (wb == null) return null;
if (sheetName == null) {
sheetName = wb.getSheetAt(0).getSheetName();
}
wb.getSheet(sheetName).shiftRows(start, end, step);
return wb;
}
public static Workbook setHeaderStyle(Workbook wb, String sheetName) {
Font font = wb.createFont();
CellStyle style = wb.createCellStyle();
font.setBoldweight(HSSFFont.BOLDWEIGHT_BOLD);
font.setFontHeightInPoints(FONT_HEIGHT_IN_POINTS);
font.setFontName("黑體");
style.setFont(font);
if (Utils.isEmpty(sheetName)) {
sheetName = wb.getSheetAt(0).getSheetName();
}
int row = wb.getSheet(sheetName).getFirstRowNum();
int cell = wb.getSheet(sheetName).getRow(row).getLastCellNum();
for (int i = 0; i < cell; i++) {
wb.getSheet(sheetName).getRow(row).getCell(i).setCellStyle(style);
}
return wb;
}
public static Workbook setHeaderOutline(Workbook wb, String sheetName, String title) {
if (wb == null) return null;
if (Utils.isEmpty(sheetName)) {
sheetName = wb.getSheetAt(0).getSheetName();
}
Header header = wb.getSheet(sheetName).getHeader();
header.setLeft(HSSFHeader.startUnderline() +
HSSFHeader.font("宋體", "Italic") +
"打雞血的口號(hào)!" + // 比如：愛我中華
HSSFHeader.endUnderline());
header.setCenter(HSSFHeader.fontSize(HEADER_FONT_SIZE) +
HSSFHeader.startDoubleUnderline() +
HSSFHeader.startBold() +
title +
HSSFHeader.endBold() +
HSSFHeader.endDoubleUnderline());
header.setRight("時(shí)間:" + HSSFHeader.date() + " " + HSSFHeader.time());
return wb;
}
public static Workbook setFooter(Workbook wb, String sheetName, String copyright) {
if (wb == null) return null;
if (Utils.isEmpty(sheetName)) {
sheetName = wb.getSheetAt(0).getSheetName();
}
Footer footer = wb.getSheet(sheetName).getFooter();
if (Utils.isEmpty(copyright)) {
copyright = "中華人民共和國"; // 版權(quán)信息，自己公司的名字或者app的名字
}
footer.setLeft("Copyright @ " + copyright);
footer.setCenter("Page:" + HSSFFooter.page() + " / " + HSSFFooter.numPages());
footer.setRight("File:" + HSSFFooter.file());
return wb;
}
public static String create(String sheetNm, String file, List header, List data, String title, String copyright) {
Workbook wb = createWorkbook(file);
if (Utils.isEmpty(sheetNm)) {
sheetNm = wb.getSheetAt(0).getSheetName();
}
setHeaderOutline(wb, sheetNm, title);
setHeader(wb, sheetNm, header);
setData(wb, sheetNm, 1, data);
setFooter(wb, sheetNm, copyright);
writeFile(wb, file);
if (wb != null) {
return file;
}
return null;
}
public static String getSystemFileCharset() {
Properties pro = System.getProperties();
return pro.getProperty("file.encoding");
}
// TODO 后面增加其他設(shè)置
}
HDFS工具類封裝
Hadoop分布式文件系統(tǒng)(HDFS)被設(shè)計(jì)成適合運(yùn)行在通用硬件(commodity hardware)上的分布式文件系統(tǒng)。它和現(xiàn)有的分布式文件系統(tǒng)有很多共同點(diǎn)。但同時(shí)，它和其他的分布式文件系統(tǒng)的區(qū)別也是很明顯的。HDFS是一個(gè)高度容錯(cuò)性的系統(tǒng)，適合部署在廉價(jià)的機(jī)器上。HDFS能提供高吞吐量的數(shù)據(jù)訪問，非常適合大規(guī)模數(shù)據(jù)集上的應(yīng)用。HDFS放寬了一部分POSIX約束，來實(shí)現(xiàn)流式讀取文件系統(tǒng)數(shù)據(jù)的目的。HDFS在最開始是作為Apache Nutch搜索引擎項(xiàng)目的基礎(chǔ)架構(gòu)而開發(fā)的。HDFS是Apache Hadoop Core項(xiàng)目的一部分。
HDFS有著高容錯(cuò)性（fault-tolerant）的特點(diǎn)，并且設(shè)計(jì)用來部署在低廉的（low-cost）硬件上。而且它提供高吞吐量（high throughput）來訪問應(yīng)用程序的數(shù)據(jù)，適合那些有著超大數(shù)據(jù)集（large data set）的應(yīng)用程序。HDFS放寬了（relax）POSIX的要求（requirements）這樣可以實(shí)現(xiàn)流的形式訪問（streaming access）文件系統(tǒng)中的數(shù)據(jù)。
在gradle中引入hdfs：
// jersey
compile "com.sun.jersey:jersey-core:1.19.1"
compile "com.sun.jersey:jersey-server:1.19.1"
compile "com.sun.jersey:jersey-client:1.19.1"
compile "com.sun.jersey:jersey-json:1.19.1"
// hadoop
compile ("org.apache.hadoop:hadoop-common:2.7.2") {
exclude(module: "jersey")
exclude(module: "contribs")
}
compile ("org.apache.hadoop:hadoop-hdfs:2.7.2") {
exclude(module: "jersey")
exclude(module: "contribs")
}
compile ("org.apache.hadoop:hadoop-client:2.7.2") {
exclude(module: "jersey")
exclude(module: "contribs")
}`
package cn.test.web.utils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.apache.poi.ss.usermodel.Workbook;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URISyntaxException;
/**
* Created with test-data
* User zhoujunwen
* Date 16/8/11
* Time 下午5:41
*/
public class HDFSUtils {
private static FileSystem fs = null;
public static FileSystem getFileSystem(Configuration conf) throws IOException,
URISyntaxException {
fs = FileSystem.get(conf);
//fs = FileSystem.newInstance(conf);
return fs;
}
/**
* 判斷路徑是否存在
*
* @param conf
* @param path
* @return
* @throws IOException
*/
public static boolean exits(Configuration conf, String path) throws IOException,
URISyntaxException {
FileSystem fs = getFileSystem(conf);
return fs.exists(new Path(path));
}
/**
* 創(chuàng)建文件
*
* @param conf
* @param filePath
* @param contents
* @throws IOException
*/
public static void createFile(Configuration conf, String filePath, byte[] contents)
throws IOException, URISyntaxException {
FileSystem fs = getFileSystem(conf);
Path path = new Path(filePath);
FSDataOutputStream outputStream = fs.create(path);
outputStream.write(contents, 0, contents.length);
outputStream.hflush();
outputStream.close();
fs.close();
}
/**
* 創(chuàng)建文件
*
* @param conf
* @param filePath
* @param fileContent
* @throws IOException
*/
public static void createFile(Configuration conf, String fileContent, String filePath)
throws IOException, URISyntaxException {
createFile(conf, filePath, fileContent.getBytes());
}
/**
* 上傳文件
*
* @param conf
* @param localFilePath
* @param remoteFilePath
* @throws IOException
*/
public static void copyFromLocalFile(Configuration conf, String localFilePath, String remoteFilePath)
throws IOException, URISyntaxException {
FileSystem fs = getFileSystem(conf);
Path localPath = new Path(localFilePath);
Path remotePath = new Path(remoteFilePath);
fs.copyFromLocalFile(true, true, localPath, remotePath);
fs.close();
}
/**
* 刪除目錄或文件
*
* @param conf
* @param remoteFilePath
* @param recursive
* @return
* @throws IOException
*/
public static boolean deleteFile(Configuration conf, String remoteFilePath, boolean recursive)
throws IOException, URISyntaxException {
FileSystem fs = getFileSystem(conf);
boolean result = fs.delete(new Path(remoteFilePath), recursive);
fs.close();
return result;
}
/**
* 刪除目錄或文件(如果有子目錄,則級(jí)聯(lián)刪除)
*
* @param conf
* @param remoteFilePath
* @return
* @throws IOException
*/
public static boolean deleteFile(Configuration conf, String remoteFilePath)
throws IOException, URISyntaxException {
return deleteFile(conf, remoteFilePath, true);
}
/**
* 文件重命名
*
* @param conf
* @param oldFileName
* @param newFileName
* @return
* @throws IOException
*/
public static boolean renameFile(Configuration conf, String oldFileName, String newFileName)
throws IOException, URISyntaxException {
FileSystem fs = getFileSystem(conf);
Path oldPath = new Path(oldFileName);
Path newPath = new Path(newFileName);
boolean result = fs.rename(oldPath, newPath);
fs.close();
return result;
}
/**
* 創(chuàng)建目錄
*
* @param conf
* @param dirName
* @return
* @throws IOException
*/
public static boolean createDirectory(Configuration conf, String dirName)
throws IOException, URISyntaxException {
FileSystem fs = getFileSystem(conf);
Path dir = new Path(dirName);
boolean result = fs.mkdirs(dir);
fs.close();
return result;
}
/**
* 列出指定路徑下的所有文件(不包含目錄)
*
* @param fs
* @param basePath
* @param recursive
*/
public static RemoteIterator listFiles(FileSystem fs, String basePath, boolean recursive)
throws IOException {
RemoteIterator fileStatusRemoteIterator = fs.listFiles(new Path(basePath), recursive);
return fileStatusRemoteIterator;
}
/**
* 列出指定路徑下的文件（非遞歸）
*
* @param conf
* @param basePath
* @return
* @throws IOException
*/
public static RemoteIterator listFiles(Configuration conf, String basePath)
throws IOException, URISyntaxException {
FileSystem fs = getFileSystem(conf);
RemoteIterator remoteIterator = fs.listFiles(new Path(basePath), false);
fs.close();
return remoteIterator;
}
/**
* 列出指定目錄下的文件子目錄信息（非遞歸）
*
* @param conf
* @param dirPath
* @return
* @throws IOException
*/
public static FileStatus[] listStatus(Configuration conf, String dirPath) throws IOException,
URISyntaxException {
FileSystem fs = getFileSystem(conf);
FileStatus[] fileStatuses = fs.listStatus(new Path(dirPath));
fs.close();
return fileStatuses;
}
/**
* 讀取文件內(nèi)容并寫入outputStream中
*
* @param conf 配置
* @param filePath 文件路徑
* @param os 輸出流
* @return
* @throws IOException
*/
public static void readFile(Configuration conf, String filePath, OutputStream os) throws IOException,
URISyntaxException {
FileSystem fs = getFileSystem(conf);
Path path = new Path(filePath);
try (FSDataInputStream inputStream = fs.open(path)) {
Workbook wb = POIUtils.createWorkbookByIS(filePath, inputStream);
wb.write(os);
inputStream.close();
} finally {
fs.close();
}
}
/**
* 讀取文件內(nèi)容并返回
* @param conf
* @param filePath
* @return
* @throws IOException
* @throws URISyntaxException
*/
public static String readFile(Configuration conf, String filePath) throws IOException,
URISyntaxException {
String fileContent = null;
FileSystem fs = getFileSystem(conf);
Path path = new Path(filePath);
InputStream inputStream = null;
ByteArrayOutputStream outputStream = null;
try {
inputStream = fs.open(path);
outputStream = new ByteArrayOutputStream(inputStream.available());
IOUtils.copyBytes(inputStream, outputStream, conf);
byte[] lens = outputStream.toByteArray();
fileContent = new String(lens, "UTF-8");
} finally {
IOUtils.closeStream(inputStream);
IOUtils.closeStream(outputStream);
fs.close();
}
return fileContent;
}
}
對(duì)于hdfs我多帶帶有謝了兩個(gè)類，一個(gè)是HDFSFileUploader,一個(gè)是Configuration。如類名，前者用于文件上傳，后者用于hdfs的配置。
HDFSFileUploaderpackage cn.test.web.utils.hadoop;
import cn.test.common.log.Log;
import cn.test.common.log.LogFactory;
import cn.test.common.util.Utils;
import cn.test.web.utils.HDFSUtils;
import org.apache.commons.lang.NullArgumentException;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.UUID;
/**
* Created with test-data
* User zhoujunwen
* Date 16/8/11
* Time 下午5:42
*/
public class HDFSFileUploader {
public static final byte FROM_LOCAL_COPY = 1; // 從本地上傳文件
public static final byte FROM_CONTENT_WRITE = 2; // 讀取字符串或字節(jié),生成文件
private static final Log LOGGER = LogFactory.getLog(HDFSFileUploader.class);
private static final String HDFS_SCHEMA = "hdfs://";
private static final String SEPARATOR = "/";
private static final String SUFFIX_PREFIX = ".";
private static final int BUFFER_SIZE = 1024;
private static final Configuration CONF = new Configuration();
/**
* 上傳二進(jìn)制文件,使用默認(rèn)配置的域名,隨機(jī)生成文件名
*
* @param path
* @param suffix
* @param contents
* @return
*/
public static String upload(String path, String suffix, byte[] contents) {
return upload(null, path, suffix, contents);
}
/**
* 上傳二進(jìn)制文件,隨機(jī)生成文件名
*
* @param domain
* @param path
* @param suffix
* @param contents
* @return
*/
public static String upload(String domain, String path, String suffix, byte[] contents) {
return upload(domain, path, null, suffix, contents);
}
/**
* 上傳二進(jìn)制文件,指定文件名,只能通過流上傳
*
* @param domain
* @param path
* @param filename
* @param suffix
* @param content
* @return
*/
public static String upload(String domain, String path, String filename, String suffix,
final byte[] content) {
return upload(domain, path, filename, suffix, new String(content), FROM_CONTENT_WRITE);
}
/**
* 上傳文件,默認(rèn)域名和隨機(jī)文件名
*
* @param path
* @param suffix
* @param src
* @return
*/
public static String upload(String path, String suffix, String src, byte fromLocal) {
return upload(null, path, suffix, src, fromLocal);
}
/**
* 上傳文件到指定域名的指定目錄,文件名隨機(jī)生成
*
* @param domain 域名,比如 10.25.126.28:9000
* @param path 文件路徑,比如 /usr/local/com.hd.test/2016-08-08/
* @param suffix 文件后綴,比如 .xsl,xsl
* @param src 文件內(nèi)容,字符串 || 本地文件路徑
* @return String 完整的文件名
*/
public static String upload(String domain, String path, String suffix, String src, byte
fromLocal) {
return upload(domain, path, null, suffix, src, fromLocal);
}
/**
* 上傳文件,指定了域名,路徑,文件名,后綴
*
* @param domain 域名
* @param path 路徑
* @param filename 文件名
* @param suffix 后綴
* @param src 內(nèi)容 || 本地路徑
* @return
*/
public static String upload(String domain, String path, String filename, String suffix, String
src, byte fromLocal) {
String filePath = getRealAddr(domain, path, suffix, filename);
System.out.println(filePath);
try {
switch (fromLocal) {
case FROM_LOCAL_COPY:
HDFSUtils.copyFromLocalFile(CONF, src, filePath);
break;
case FROM_CONTENT_WRITE:
HDFSUtils.createFile(CONF, src, filePath);
break;
}
return filePath;
} catch (IOException | URISyntaxException e) {
LOGGER.warn("上傳文件失敗:{}",e.getMessage());
}
return null;
}
/**
* 文件完整的路徑
*
* @param domain 域名
* @param path 目錄路徑
* @param suffix 后綴
* @param filename 文件名
* @return
*/
private static String getRealAddr(String domain, String path, String suffix, String filename) {
if (!Utils.isEmpty(domain) && !domain.startsWith(HDFS_SCHEMA)) {
domain = HDFS_SCHEMA + domain;
} else {
domain = "";
}
path = getPath(path);
filename = getFilename(filename, suffix);
return String.format("%s%s%s", domain, path, filename);
}
/**
* 文件路徑
*
* @param path
* @return
*/
private static String getPath(String path) {
if (Utils.isEmpty(path)) {
throw new NullArgumentException("path id null");
}
if (!path.startsWith(SEPARATOR)) {
path = SEPARATOR + path;
}
if (!path.endsWith(SEPARATOR)) {
path = path + SEPARATOR;
}
return path;
}
/**
* 生成文件名
*
* @param filename
* @param suffix
* @return
*/
private static String getFilename(String filename, String suffix) {
if (Utils.isEmpty(filename)) {
filename = generateFilename();
}
if (!Utils.isEmpty(suffix)) {
filename = suffix.equals(SEPARATOR) ? filename : (filename.endsWith(suffix) ?
filename : ((filename.endsWith(SUFFIX_PREFIX)
|| suffix.startsWith(SUFFIX_PREFIX)) ? filename + suffix
: filename + SUFFIX_PREFIX + suffix));
}
return filename;
}
/**
* 生成文件名
*
* @return
*/
private static String generateFilename() {
return getUuid(false);
}
/**
* 生成UUID
*
* @param isNeedHyphen
* @return
*/
public static String getUuid(boolean isNeedHyphen) {
UUID uuid = UUID.randomUUID();
String str = uuid.toString();
if (isNeedHyphen) {
str = str.replaceAll("-", "");
}
return str;
}
public static void setConfResource(final Configuration config) {
CONF.addResource(config);
}
}
HDFSFileUploader中的一系列方法，用于上傳不同類型的文件，比如二進(jìn)制文件，字符串等，還有hdfs的copy本地文件以及文件名uuid生成等方法。
Configurationpackage cn.test.web.utils.hadoop; import cn.test.web.utils.CommonUtils; import org.apache.commons.io.FilenameUtils; import org.springframework.core.io.Resource; import org.springframework.core.io.support.PathMatchingResourcePatternResolver; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; /** * Created with test-data * User zhoujunwen * Date 16/8/9 * Time 上午9:30 * 建議使用方法: ** * 在使用的地方直接注入hadoopConfig: * * @Resource private Configuration hadoopConfig; */ public class Configuration extends org.apache.hadoop.conf.Configuration { private Resource[] resources; public void setResources(List* **
*classpath:/spring/core-site.xml *filenames) throws IOException { List resources = new ArrayList<>(); if (filenames != null && filenames.size() > 0) { for (String filename : filenames) { filename = filename.trim(); String realName = getFileName(filename); String ext = FilenameUtils.getExtension(realName); if (ext.equals("xml")) { PathMatchingResourcePatternResolver pathMatchingResourcePatternResolver = new PathMatchingResourcePatternResolver(); try { Resource[] resourceList = pathMatchingResourcePatternResolver.getResources(filename); Collections.addAll(resources, resourceList); } catch (IOException e) { e.printStackTrace(); } } } } for (Resource resource : resources) { this.addResource(resource.getURL()); } } private String getFileName(String fileName) { return CommonUtils.getFileName(fileName); } }
這個(gè)類很簡單，其實(shí)是集成了hadoop的org.apache.hadoop.conf.Configuration類，目的是為了在spring配置文件中，靈活的指定hadoop的配置文件，所用到的就是org.apache.hadoop.conf.Configuration的addResource(String name)方法,下面是在spring xml中的配置。
classpath:META-INF/hadoop/*.xml
導(dǎo)出訂單處理（mq消費(fèi)端）
