乐趣区

聊聊puma的DefaultDataHandler

jiezi

序

本文主要研究一下 puma 的 DefaultDataHandler

DataHandler

puma/puma/src/main/java/com/dianping/puma/datahandler/DataHandler.java

public interface DataHandler extends LifeCycle {DataHandlerResult process(BinlogEvent binlogEvent, PumaContext context);
}
  • DataHandler 继承了 LifeCycle，它定义了 process 方法

AbstractDataHandler

puma/puma/src/main/java/com/dianping/puma/datahandler/AbstractDataHandler.java

@ThreadUnSafe
public abstract class AbstractDataHandler implements DataHandler {private static final Logger log = Logger.getLogger(AbstractDataHandler.class);

    private TableMetaInfoFetcher tableMetasInfoFetcher;

    /**
     * @return the tableMetasInfoFetcher
     */
    public TableMetaInfoFetcher getTableMetasInfoFetcher() {return tableMetasInfoFetcher;}

    /**
     * @param tableMetasInfoFetcher the tableMetasInfoFetcher to set
     */
    public void setTableMetasInfoFetcher(TableMetaInfoFetcher tableMetasInfoFetcher) {this.tableMetasInfoFetcher = tableMetasInfoFetcher;}

    @Override
    public void start() {}


    @Override
    public void stop() {}

    protected Object convertUnsignedValueIfNeeded(int pos, Object value, TableMetaInfo tableMeta) {
        Object newValue = value;
        if (value != null) {switch (tableMeta.getRawTypeCodes().get(pos)) {
                case BinlogConstants.MYSQL_TYPE_TINY:
                    if ((value instanceof Integer) && (Integer) value < 0 && !tableMeta.getSignedInfos().get(pos)) {newValue = Integer.valueOf((Integer) value + (1 << 8));
                    }
                    break;
                case BinlogConstants.MYSQL_TYPE_INT24:
                    if ((value instanceof Integer) && (Integer) value < 0 && !tableMeta.getSignedInfos().get(pos)) {newValue = Integer.valueOf((Integer) value + (1 << 24));
                    }
                    break;
                case BinlogConstants.MYSQL_TYPE_SHORT:
                    if ((value instanceof Integer) && (Integer) value < 0 && !tableMeta.getSignedInfos().get(pos)) {newValue = Integer.valueOf((Integer) value + (1 << 16));
                    }
                    break;
                case BinlogConstants.MYSQL_TYPE_INT:
                    if ((value instanceof Integer) && (Integer) value < 0 && !tableMeta.getSignedInfos().get(pos)) {newValue = Long.valueOf((Integer) value) + (1L << 32);
                    } else {if (value instanceof Integer) {newValue = Long.valueOf((Integer) value);
                        }
                    }
                    break;
                case BinlogConstants.MYSQL_TYPE_LONGLONG:
                    if ((value instanceof Long) && (Long) value < 0 && !tableMeta.getSignedInfos().get(pos)) {newValue = BigInteger.valueOf((Long) value).add(BigInteger.ONE.shiftLeft(64));
                    } else {if (value instanceof Long) {newValue = BigInteger.valueOf((Long) value);
                        }
                    }
                    break;
                default:
                    break;
            }

        }
        return newValue;
    }

    @Override
    public DataHandlerResult process(BinlogEvent binlogEvent, PumaContext context) {DataHandlerResult result = new DataHandlerResult();
        if (binlogEvent instanceof PumaIgnoreEvent) {log.info("Ingore one unknown event. eventType:" + binlogEvent.getHeader().getEventType());
            result.setEmpty(true);
            result.setFinished(true);
            return result;
        }

        byte eventType = binlogEvent.getHeader().getEventType();
        if (log.isDebugEnabled()) {log.debug("event#" + eventType);
        }
        if (eventType == BinlogConstants.STOP_EVENT || eventType == BinlogConstants.ROTATE_EVENT) {result.setEmpty(true);
            result.setFinished(true);
        } else if (eventType == BinlogConstants.FORMAT_DESCRIPTION_EVENT) {result.setEmpty(true);
            result.setFinished(true);
        } else if (eventType == BinlogConstants.QUERY_EVENT) {handleQueryEvent(binlogEvent, result);
        } else {doProcess(result, binlogEvent, context, eventType);
        }

        if (result != null && !result.isEmpty() && result.getData() != null) {BinlogInfo binlogInfo = new BinlogInfo(context.getDBServerId(), context.getBinlogFileName(),
                    context.getBinlogStartPos(), context.getEventIndex(), binlogEvent.getHeader().getTimestamp());
            result.getData().setBinlogInfo(binlogInfo);
            result.getData().setServerId(binlogEvent.getHeader().getServerId());
        }

        return result;
    }

    protected void handleQueryEvent(BinlogEvent binlogEvent, DataHandlerResult result) {QueryEvent queryEvent = (QueryEvent) binlogEvent;
        String sql = StringUtils.normalizeSpace(queryEvent.getSql());
        if (StringUtils.startsWithIgnoreCase(sql, "ALTER") || StringUtils.startsWithIgnoreCase(sql, "CREATE")
                || StringUtils.startsWithIgnoreCase(sql, "DROP") || StringUtils.startsWithIgnoreCase(sql, "RENAME")
                || StringUtils.startsWithIgnoreCase(sql, "TRUNCATE")) {handleDDlEvent(result, queryEvent, sql);

        } else if (StringUtils.equalsIgnoreCase(sql, "BEGIN")) {handleTransactionBeginEvent(binlogEvent, result, queryEvent);
        } else {result.setEmpty(true);
            result.setFinished(true);
            // log.info("QueryEvent  sql=" + queryEvent.getSql());
        }
    }

    protected void handleTransactionBeginEvent(BinlogEvent binlogEvent, DataHandlerResult result,
                                               QueryEvent queryEvent) {
        // BEGIN 事件，发送一个 begin transaction 的事件
        ChangedEvent dataChangedEvent = new RowChangedEvent();
        ((RowChangedEvent) dataChangedEvent).setTransactionBegin(true);
        dataChangedEvent.setExecuteTime(binlogEvent.getHeader().getTimestamp());
        dataChangedEvent.setDatabase(queryEvent.getDatabaseName());

        result.setData(dataChangedEvent);
        result.setEmpty(false);
        result.setFinished(true);
    }

    /**
     * @param result
     * @param queryEvent
     * @param sql
     */
    protected void handleDDlEvent(DataHandlerResult result, QueryEvent queryEvent, String sql) {ChangedEvent dataChangedEvent = new DdlEvent();
        DdlEvent ddlEvent = (DdlEvent) dataChangedEvent;
        ddlEvent.setSql(sql);
        ddlEvent.setDdlEventType(SimpleDdlParser.getEventType(sql));

        ddlEvent.setDdlEventSubType(SimpleDdlParser.getEventSubType(ddlEvent.getDdlEventType(), sql));
        if (ddlEvent.getDdlEventType() == DdlEventType.DDL_DEFAULT
                || ddlEvent.getDdlEventSubType() == DdlEventSubType.DDL_SUB_DEFAULT) {log.info("DdlEvent Type do not found. ddl sql=" + sql);
        }
        SimpleDdlParser.DdlResult ddlResult = SimpleDdlParser
                .getDdlResult(ddlEvent.getDdlEventType(), ddlEvent.getDdlEventSubType(), sql);
        if (ddlResult != null) {ddlEvent.setDatabase(StringUtils.isNotBlank(ddlResult.getDatabase()) ? ddlResult.getDatabase()
                    : StringUtils.EMPTY);
            ddlEvent.setTable(StringUtils.isNotBlank(ddlResult.getTable()) ? ddlResult.getTable() : StringUtils.EMPTY);
            if (ddlEvent.getDdlEventType() != DdlEventType.DDL_CREATE) {log.info("DDL event, sql=" + sql + ",database =" + ddlResult.getDatabase() + "table ="
                        + ddlResult.getTable() + "queryEvent.getDatabaseName()" + queryEvent.getDatabaseName());
            }
        }
        if (StringUtils.isBlank(ddlEvent.getDatabase())) {ddlEvent.setDatabase(queryEvent.getDatabaseName());
        }

        if (ddlEvent.getDdlEventType() == DdlEventType.DDL_ALTER
                && ddlEvent.getDdlEventSubType() == DdlEventSubType.DDL_ALTER_TABLE) {ddlEvent.setDDLType(DDLType.ALTER_TABLE);
        }

        tableMetasInfoFetcher.refreshTableMeta(ddlEvent.getDatabase(), ddlEvent.getTable());

        ddlEvent.setExecuteTime(queryEvent.getHeader().getTimestamp());
        result.setData(dataChangedEvent);
        result.setEmpty(false);
        result.setFinished(true);
    }

    protected abstract void doProcess(DataHandlerResult result, BinlogEvent binlogEvent, PumaContext context,
                                      byte eventType);
}
  • AbstractDataHandler 声明实现了 DataHandler 接口，其 process 方法针对 BinlogConstants.QUERY_EVENT 执行 handleQueryEvent，非 STOP_EVENT、ROTATE_EVENT、FORMAT_DESCRIPTION_EVENT 的 event 执行 doProcess，该方法由子类去实现；handleQueryEvent 方法针对 ALTER、CREATE、DROP、RENAME、TRUNCATE 的执行 handleDDlEvent 方法，针对 BEGIN 的执行 handleTransactionBeginEvent 方法；handleDDlEvent 方法会执行 tableMetasInfoFetcher.refreshTableMeta；handleTransactionBeginEvent 方法标记 result 的 finished 为 true

DefaultDataHandler

puma/puma/src/main/java/com/dianping/puma/datahandler/DefaultDataHandler.java

@ThreadUnSafe
public class DefaultDataHandler extends AbstractDataHandler {private Logger log = Logger.getLogger(DefaultDataHandler.class);

    private Map<Long, TableMetaInfo> tableMetaInfos;

    private int rowPos = 0;

    @Override
    protected void doProcess(DataHandlerResult result, BinlogEvent binlogEvent, PumaContext context, byte eventType) {if (log.isDebugEnabled()) {log.debug("event:" + eventType);
        }
        switch (eventType) {
            case BinlogConstants.TABLE_MAP_EVENT:
                TableMapEvent tableMapEvent = (TableMapEvent) binlogEvent;

                if (tableMetaInfos == null) {tableMetaInfos = new HashMap<Long, TableMetaInfo>();
                }

                TableMetaInfo tableMetaInfo = getTableMetasInfoFetcher().getTableMetaInfo(tableMapEvent.getDatabaseName(),
                        tableMapEvent.getTableName());

                if (tableMetaInfo != null) {tableMetaInfos.put(tableMapEvent.getTableId(), tableMetaInfo);
                    if (log.isDebugEnabled()) {log.debug("put meta info for table id:" + tableMapEvent.getTableId());
                    }
                } else {if (log.isDebugEnabled()) {log.debug("meta info not found for:" + tableMapEvent.getDatabaseName() + "-"
                                + tableMapEvent.getTableName());
                    }
                    skipEvent(BinlogConstants.TABLE_MAP_EVENT, result, context);
                    return;
                }

                fillRawTypeCodes(tableMapEvent, tableMetaInfo);
                fillRawNullAbilities(tableMapEvent, tableMetaInfo);
                rowPos = 0;
                result.setEmpty(true);
                result.setFinished(true);
                break;
            case BinlogConstants.WRITE_ROWS_EVENT_V1:
            case BinlogConstants.WRITE_ROWS_EVENT:
                if (tableMetaInfos == null || tableMetaInfos.isEmpty()) {skipEvent(BinlogConstants.WRITE_ROWS_EVENT, result, context);
                    return;
                }

                processWriteRowEvent(result, binlogEvent, context);
                break;
            case BinlogConstants.UPDATE_ROWS_EVENT_V1:
            case BinlogConstants.UPDATE_ROWS_EVENT:
                if (tableMetaInfos == null || tableMetaInfos.isEmpty()) {skipEvent(BinlogConstants.UPDATE_ROWS_EVENT, result, context);
                    return;
                }
                processUpdateRowEvent(result, binlogEvent, context);
                break;
            case BinlogConstants.DELETE_ROWS_EVENT_V1:
            case BinlogConstants.DELETE_ROWS_EVENT:
                if (tableMetaInfos == null || tableMetaInfos.isEmpty()) {skipEvent(BinlogConstants.DELETE_ROWS_EVENT, result, context);
                    return;
                }
                processDeleteRowEvent(result, binlogEvent, context);
                break;
            case BinlogConstants.XID_EVENT:
                if (tableMetaInfos == null || tableMetaInfos.isEmpty()) {skipEvent(BinlogConstants.XID_EVENT, result, context);
                    return;
                }
                processTransactionCommitEvent(binlogEvent, result);
                break;
            default:
                result.setEmpty(true);
                result.setFinished(true);
                break;
        }

    }

    //......

}
  • DefaultDataHandler 继承了 AbstractDataHandler，其 doProcess 方法针对不同的 eventType 做不同的处理；对于 TABLE_MAP_EVENT 更新 tableMetaInfo；对于 WRITE_ROWS_EVENT_V1、WRITE_ROWS_EVENT 执行 processWriteRowEvent；对于 UPDATE_ROWS_EVENT_V1、UPDATE_ROWS_EVENT 执行 processUpdateRowEvent；对于 DELETE_ROWS_EVENT_V1、DELETE_ROWS_EVENT 执行 processDeleteRowEvent；对于 XID_EVENT 执行 processTransactionCommitEvent

processWriteRowEvent

    protected void processWriteRowEvent(DataHandlerResult result, BinlogEvent binlogEvent, PumaContext context) {WriteRowsEvent writeRowsEvent = (WriteRowsEvent) binlogEvent;

        if (rowPos >= writeRowsEvent.getRows().size()) {
            rowPos = 0;
            result.setEmpty(true);
            result.setFinished(true);
        } else {TableMetaInfo tableMetaInfo = tableMetaInfos.get(writeRowsEvent.getTableId());

            if (tableMetaInfo == null) {skipEvent(BinlogConstants.WRITE_ROWS_EVENT, result, context);
                return;
            }

            RowChangedEvent rowChangedEvent = new RowChangedEvent();
            Map<String, ColumnInfo> columns = initColumns(writeRowsEvent, rowChangedEvent, DMLType.INSERT,
                    tableMetaInfo);

            for (int columnPos = 0, columnIndex = 0; columnPos < writeRowsEvent.getColumnCount().intValue(); columnPos++) {if (writeRowsEvent.getUsedColumns().get(columnPos)) {Column binlogColumn = writeRowsEvent.getRows().get(rowPos).getColumns().get(columnIndex);
                    String columnName = tableMetaInfo.getColumns().get(columnPos + 1);
                    if (!checkUnknownColumnName(result, context, columnName, columnPos + 1)) {return;}
                    ColumnInfo columnInfo = new ColumnInfo(tableMetaInfo.getKeys().contains(columnName), null,
                            convertUnsignedValueIfNeeded(columnPos + 1, binlogColumn.getValue(), tableMetaInfo));
                    columns.put(columnName, columnInfo);
                    columnIndex++;
                }
            }

            rowPos++;
            result.setData(rowChangedEvent);
            result.setEmpty(false);
            result.setFinished(false);
        }
    }
  • processWriteRowEvent 方法主要构建 rowChangedEvent，并填充其 columns

processUpdateRowEvent

    protected void processUpdateRowEvent(DataHandlerResult result, BinlogEvent binlogEvent, PumaContext context) {UpdateRowsEvent updateRowsEvent = (UpdateRowsEvent) binlogEvent;

        if (rowPos >= updateRowsEvent.getRows().size()) {
            rowPos = 0;
            result.setEmpty(true);
            result.setFinished(true);
        } else {TableMetaInfo tableMetaInfo = tableMetaInfos.get(updateRowsEvent.getTableId());

            if (tableMetaInfo == null) {skipEvent(BinlogConstants.UPDATE_ROWS_EVENT, result, context);
                return;
            }

            RowChangedEvent rowChangedEvent = new RowChangedEvent();
            Map<String, ColumnInfo> columns = initColumns(updateRowsEvent, rowChangedEvent, DMLType.UPDATE,
                    tableMetaInfo);
            if (log.isDebugEnabled()) {log.debug("update from" + tableMetaInfo.getDatabase() + "." + tableMetaInfo.getTable());
            }

            for (int columnPos = 0, columnAfterIndex = 0, columnBeforeIndex = 0; columnPos < updateRowsEvent
                    .getColumnCount().intValue(); columnPos++) {String columnName = tableMetaInfo.getColumns().get(columnPos + 1);
                if (!checkUnknownColumnName(result, context, columnName, columnPos + 1)) {return;}
                Column afterColumn = null;
                Column beforeColumn = null;
                if (updateRowsEvent.getUsedColumnsAfter().get(columnPos)) {afterColumn = updateRowsEvent.getRows().get(rowPos).getAfter().getColumns().get(columnAfterIndex);
                    columnAfterIndex++;
                }
                if (updateRowsEvent.getUsedColumnsBefore().get(columnPos)) {beforeColumn = updateRowsEvent.getRows().get(rowPos).getBefore().getColumns().get(columnBeforeIndex);
                    columnBeforeIndex++;
                }
                ColumnInfo columnInfo = new ColumnInfo(tableMetaInfo.getKeys().contains(columnName),
                        beforeColumn == null ? null : convertUnsignedValueIfNeeded(columnPos + 1, beforeColumn.getValue(),
                                tableMetaInfo), afterColumn == null ? null : convertUnsignedValueIfNeeded(columnPos + 1,
                        afterColumn.getValue(), tableMetaInfo));
                columns.put(columnName, columnInfo);
            }

            rowPos++;
            result.setData(rowChangedEvent);
            result.setEmpty(false);
            result.setFinished(false);
        }
    }
  • processUpdateRowEvent 主要构建 rowChangedEvent，并填充其 columns

processDeleteRowEvent

    protected void processDeleteRowEvent(DataHandlerResult result, BinlogEvent binlogEvent, PumaContext context) {DeleteRowsEvent deleteRowsEvent = (DeleteRowsEvent) binlogEvent;

        if (rowPos >= deleteRowsEvent.getRows().size()) {
            rowPos = 0;
            result.setEmpty(true);
            result.setFinished(true);
        } else {TableMetaInfo tableMetaInfo = tableMetaInfos.get(deleteRowsEvent.getTableId());

            if (tableMetaInfo == null) {skipEvent(BinlogConstants.DELETE_ROWS_EVENT, result, context);
                return;
            }

            RowChangedEvent rowChangedEvent = new RowChangedEvent();
            Map<String, ColumnInfo> columns = initColumns(deleteRowsEvent, rowChangedEvent, DMLType.DELETE,
                    tableMetaInfo);

            for (int columnPos = 0, columnIndex = 0; columnPos < deleteRowsEvent.getColumnCount().intValue(); columnPos++) {if (deleteRowsEvent.getUsedColumns().get(columnPos)) {Column binlogColumn = deleteRowsEvent.getRows().get(rowPos).getColumns().get(columnIndex);
                    String columnName = tableMetaInfo.getColumns().get(columnPos + 1);
                    if (!checkUnknownColumnName(result, context, columnName, columnPos + 1)) {return;}
                    ColumnInfo columnInfo = new ColumnInfo(tableMetaInfo.getKeys().contains(columnName),
                            convertUnsignedValueIfNeeded(columnPos + 1, binlogColumn.getValue(), tableMetaInfo), null);
                    columns.put(columnName, columnInfo);
                    columnIndex++;
                }
            }

            rowPos++;
            result.setData(rowChangedEvent);
            result.setEmpty(false);
            result.setFinished(false);
        }
    }
  • processDeleteRowEvent 主要构建 rowChangedEvent，并填充其 columns

processTransactionCommitEvent

    protected void processTransactionCommitEvent(BinlogEvent binlogEvent, DataHandlerResult result) {
        // commit 事件，发送一个 commit transaction 的事件
        ChangedEvent dataChangedEvent = new RowChangedEvent();
        ((RowChangedEvent) dataChangedEvent).setTransactionCommit(true);
        dataChangedEvent.setExecuteTime(binlogEvent.getHeader().getTimestamp());
        dataChangedEvent.setDatabase(tableMetaInfos.values().iterator().next().getDatabase());

        result.setData(dataChangedEvent);
        result.setEmpty(false);
        result.setFinished(true);
        tableMetaInfos.clear();
        tableMetaInfos = null;
    }
  • processTransactionCommitEvent 方法主要是构建 dataChangedEvent，并标记 result 的 finished 为 true，并清空 tableMetaInfos

小结

DefaultDataHandler 继承了 AbstractDataHandler，其 doProcess 方法针对不同的 eventType 做不同的处理；对于 TABLE_MAP_EVENT 更新 tableMetaInfo；对于 WRITE_ROWS_EVENT_V1、WRITE_ROWS_EVENT 执行 processWriteRowEvent；对于 UPDATE_ROWS_EVENT_V1、UPDATE_ROWS_EVENT 执行 processUpdateRowEvent；对于 DELETE_ROWS_EVENT_V1、DELETE_ROWS_EVENT 执行 processDeleteRowEvent；对于 XID_EVENT 执行 processTransactionCommitEvent

doc

  • DefaultDataHandler
退出移动版