Pulsar集群 监听mysql8的binlog java代码

package com.chen.delerium.utils; import com.chen.delerium.model.OriginExcelModel; import org.apache.poi.hssf.usermodel.HSSFWorkbook; import org.apache.poi.ss.usermodel.*; import org.apache.poi.xssf.usermodel.XSSFWorkbook; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.*; import java.util.ArrayList; import java.util.List; public class ExcelReader { private static Logger logger = LoggerFactory.getLogger(ExcelReader.class); private static final String XLS = "xls"; private static final String XLSX = "xlsx"; /** * 根据文件后缀名类型获取对应的工作簿对象 * @param inputStream 读取文件的输入流 * @param fileType 文件后缀名类型（xls或xlsx） * @return 包含文件数据的工作簿对象 * @throws IOException */ public static Workbook getWorkbook(InputStream inputStream, String fileType) throws IOException { Workbook workbook = null; if (fileType.equalsIgnoreCase(XLS)) { workbook = new HSSFWorkbook(inputStream); } else if (fileType.equalsIgnoreCase(XLSX)) { workbook = new XSSFWorkbook(inputStream); } return workbook; } /** * 读取Excel文件内容 * @return 读取结果列表，读取失败时返回null */ public static List readExcel(InputStream inputStream,String fileType,String date) { Workbook workbook = null; try { workbook = getWorkbook(inputStream, fileType); // 读取excel中的数据 List resultDataList = parseExcel(workbook,date); return resultDataList; } catch (Exception e) { logger.warn("解析Excel失败，文件名：" + " 错误信息：" + e.getMessage()); return null; } finally { try { if (null != workbook) { workbook.close(); } if (null != inputStream) { inputStream.close(); } } catch (Exception e) { logger.warn("关闭数据流出错！错误信息：" + e.getMessage()); return null; } } } /** * 解析Excel数据 * @param workbook Excel工作簿对象 * @return 解析结果 */ private static List parseExcel(Workbook workbook,String date) { List resultDataList = new ArrayList<>(); // 解析sheet for (int sheetNum = 0; sheetNum < workbook.getNumberOfSheets(); sheetNum++) { Sheet sheet = workbook.getSheetAt(sheetNum); // 校验sheet是否合法 if (sheet == null) { continue; } // 获取第一行数据 int firstRowNum = sheet.getFirstRowNum(); Row firstRow = sheet.getRow(firstRowNum); if (null == firstRow) { logger.warn("解析Excel失败，在第一行没有读取到任何数据！"); } // 解析每一行的数据，构造数据对象 int rowStart = firstRowNum + 1; int rowEnd = sheet.getPhysicalNumberOfRows(); for (int rowNum = rowStart; rowNum < rowEnd; rowNum++) { Row row = sheet.getRow(rowNum); if (null == row) { continue; } int physicalNumberOfCells = row.getPhysicalNumberOfCells(); if (physicalNumberOfCells < 3) { continue; } OriginExcelModel resultData = convertRowToData(row,date); if (null == resultData) { logger.warn("第 " + row.getRowNum() + "行数据不合法，已忽略！"); continue; } resultDataList.add(resultData); } } return resultDataList; } /** * 提取每一行中需要的数据，构造成为一个结果数据对象 * * 当该行中有单元格的数据为空或不合法时，忽略该行的数据 * * @param row 行数据 * @return 解析后的行数据对象，行数据错误时返回null */ private static OriginExcelModel convertRowToData(Row row,String date) { OriginExcelModel resultData = new OriginExcelModel(); Cell cell; int cellNum = 0; // 获取股票代碼 cell = row.getCell(cellNum++); // String code = convertCellValueToString(cell); if (null == cell) { return null; } String code = cell.getStringCellValue(); try { Integer.parseInt(code); } catch (NumberFormatException e) { // System.out.println("轉化code錯誤 -- " + code); return null; } resultData.setCode(code); // 获取name cell = row.getCell(cellNum++); String name = cell.getStringCellValue(); resultData.setName(name); // 获取內容 cell = row.getCell(cellNum++); String desc = cell.getStringCellValue(); resultData.setDesc(desc); //獲取時間 int rowNum = row.getPhysicalNumberOfCells(); String time = ""; if (rowNum > 3) { cell = row.getCell(cellNum++); time = cell.getStringCellValue(); } else { time = date; // time = DateUtil.getDate(-1,"yyyy-MM-dd"); // SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd"); // Calendar calendar = Calendar.getInstance(); // calendar.add(Calendar.DATE, -1); // Date time1 = calendar.getTime(); // time = df.format(time1); } resultData.setDate(time); return resultData; } /** * CSV文件生成方法 * @param head 文件头 * @param dataList 数据列表 * @param outPutPath 文件输出路径 * @param filename 文件名 * @return */ public static File createCSVFile(List<Object> head, List<List<Object>> dataList, String outPutPath, String filename) { File csvFile = null; BufferedWriter csvWtriter = null; try { csvFile = new File(outPutPath + File.separator + filename + ".csv"); File parent = csvFile.getParentFile(); if (parent != null && !parent.exists()) { parent.mkdirs(); } csvFile.createNewFile(); // GB2312使正确读取分隔符"," csvWtriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream( csvFile), "GB2312"), 1024); // 写入文件头部 writeRow(head, csvWtriter); // 写入文件内容 for (List<Object> row : dataList) { writeRow(row, csvWtriter); } csvWtriter.flush(); } catch (Exception e) { e.printStackTrace(); } finally { try { csvWtriter.close(); } catch (IOException e) { e.printStackTrace(); } } return csvFile; } /** * 写一行数据方法 * @param row * @param csvWriter * @throws IOException */ private static void writeRow(List<Object> row, BufferedWriter csvWriter) throws IOException { // 写入文件头部 for (Object data : row) { StringBuffer sb = new StringBuffer(); String rowStr = sb.append("\"").append(data).append("\",").toString(); csvWriter.write(rowStr); } csvWriter.newLine(); } }