@Data
// @Table 注解用來描述 JmltModel 和哪個表是關(guān)聯(lián) 的
@Table(name = "user_info") 
public class JmltModel implements Serializable {
    // @Colum 注解用來描述屬性是和哪個字段關(guān)聯(lián)的
    @Colum(name = "id")
    private Long id;
    @Colum(name = "email")
    private String email;
    @Colum(name = "name")
    private String name;
    @Colum(name = "create_time")
    private Date create_time;
}

// 1、originClass 是原始庫對象，這里通過反射獲取 Table 注解，從而拿到表名
Table table = (Table) originClass.getDeclaredAnnotation(Table.class);
String tableName = table.name();
// 2、計算所有的條數(shù)，然后按照分頁的方式進行 fetch
SqlTemplate sqlTemplate = new SqlTemplate(this.originDataSource);
int totalCount = sqlTemplate.count();
RowBounds rowBounds = new RowBounds(totalCount);
int totalPage = rowBounds.getTotalPage();
// 3、這里是按分頁批量拉取
for (int i = 1; i <= totalPage; i++) {
    int offset = rowBounds.getOffset(i);
    String condition = " limit " + offset + "," + rowBounds.getPageSize();
    ResultSet resultSet = sqlTemplate.select(condition);
    // 4、將 ResultSet 轉(zhuǎn)成 A 這里就是從 A_DB 到 A 的過程
    ResultSetExtractor<R> extractor = (ResultSetExtractor<R>) new ResultSetExtractor<>(originClass);
    List<R> result = extractor.extractData(resultSet);
    // 5、將 A 轉(zhuǎn)成 B
    List<T> targetResult = convertor.batchConvertFrom(result);
    // 6、丟到隊列中去等待消費
    this.rowObjectManager.pushToQueue(targetResult);
}

String selectSql = "SELECT * FROM employees"; 
try (ResultSet resultSet = stmt.executeQuery(selectSql)) { 
List<Employee> employees = new ArrayList<>(); 
while (resultSet.next()) 
{ 
    Employee emp = new Employee(); 
    emp.setId(resultSet.getInt("emp_id"));
    emp.setName(resultSet.getString("name"));
    emp.setPosition(resultSet.getString("position")); 
    emp.setSalary(resultSet.getDouble("salary")); 
    employees.add(emp); 
} 

/**
 * 將 resultSet 轉(zhuǎn)成 Map
 *
 * @param resultSet
 * @return
 * @throws SQLException
 */
private Map<String, Object> resultSetToMap(ResultSet resultSet) throws Exception {
    Map<String, Object> resultMap = new HashMap<>();
    // 獲取 ResultSet 的元數(shù)據(jù)
    int columnCount = resultSet.getMetaData().getColumnCount();
    // 遍歷每一列，將列名和值存儲到 Map 中
    for (int i = 1; i <= columnCount; i++) {
        String columnName = resultSet.getMetaData().getColumnName(i);
        Object value = resultSet.getObject(i);
        resultMap.put(columnName, value);
    }
    return resultMap;
}

private T mapResultSetToObject(Map<String, Object> resultMap, Class<T> objectType) throws Exception {
    // 通過目標(biāo)對象類型構(gòu)建一個對象
    T object = objectType.newInstance();  
    // 將 map 的 key 作為 field 的名字，map 的 value 作為 field 的值
    for (Map.Entry<String, Object> entry : resultMap.entrySet()) {  
        String fieldName = entry.getKey();  
        Object value = entry.getValue();  
        try {  
            Field declaredField = objectType.getDeclaredField(fieldName);  
            declaredField.setAccessible(true);  
            declaredField.set(object, value);  
        } catch (NoSuchFieldException e) {  
            LOGGER.error("ignore exception, fieldName: " + fieldName + ", objectType: " + objectType);  
        }  
    }
    // 完成對象的填充并返回
    return object;  
}

// T 是目標(biāo)對象類型，R 是原始對象類型
public interface Convertor<T, R> {  
/**  
* 將 T 轉(zhuǎn)換成 R  
*  
* @param origin  
* @return  
*/  
T convertFrom(R origin) throws SQLException;  
/**  
* 將 R 轉(zhuǎn)換成 T  
*  
* @param target  
* @return  
*/  
R convertTo(T target);  
/**  
* 批量轉(zhuǎn)換  
* @param origin  
* @return  
* @throws SQLException  
*/  
List<T> batchConvertFrom(List<R> origin) throws SQLException;  
/**  
* 批量轉(zhuǎn)換將 R 轉(zhuǎn)換成 T  
*  
* @param target  
* @return  
*/ 
List<R> batchConvertTo(List<T> target);

public void onEvent(RowObjectEvent rowObjectEvent, long sequence, boolean endOfBatch) {  
    // targetResult  
    List<T> targetResult = (List) rowObjectEvent.getRowObject();  
    // 將 T 寫到 目標(biāo)庫  
    Connection tc = null;  
    PreparedStatement pstm = null;  
    try {  
        // 下面即為獲取連接，創(chuàng)建 prepareStatement 和執(zhí)行
        tc = this.targetDataSource.getConnection();  
        SqlTemplate<T> sqlTemplate = new SqlTemplate<>(this.targetDataSource); 
        sqlTemplate.setObj(targetResult.get(0));  
        String sql = sqlTemplate.createBaseSql();  
        pstm = tc.prepareStatement(sql);  
        for (T item : targetResult) {  
            Object[] objects = sqlTemplate.createInsertSql(item);  
            for (int i = 1; i <= objects.length; i++) {  
                if (objects[i - 1] instanceof Long) {  
                    objects[i - 1] = (Long) objects[i - 1] + 1;  
                }  
                pstm.setObject(i, objects[i - 1]);  
            } 
            // 這里執(zhí)行時批量操作的
            pstm.addBatch();  
        }  
        pstm.executeUpdate();  
    } catch (Exception e) {  
        // ignore some code...
    }  
}