canal 客户端api 使用说明

谁践踏了优雅 2022-10-11 01:23 534阅读 0赞

--------------------

## canal 客户端api 使用说明 ##

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**导入 jar 包**

<dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.server</artifactId>
        <version>1.1.5</version>
    </dependency>
    
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.5</version>
    </dependency>

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**相关类与接口**

**ClientIdentity：客户端标识**

public class ClientIdentity implements Serializable {
        private static final long serialVersionUID = -8262100681930834834L;
        private String destination;
        private short clientId;
        private String filter;
    
        public ClientIdentity() {
        public ClientIdentity(String destination, short clientId) {
        public ClientIdentity(String destination, short clientId, String filter) {
    
    
        public Boolean hasFilter() {
            return this.filter == null ? false : StringUtils.isNotBlank(this.filter);
        }
    
    
        public String getFilter() {
        public short getClientId() {
        public String getDestination() {
    
        public void setFilter(String filter) {
        public void setClientId(short clientId) {
        public void setDestination(String destination) {
    
    
        public String toString() {
        public int hashCode() {
        public boolean equals(Object obj) {

**CanalConnector：客户端连接接口**

public interface CanalConnector {
    
        void connect() throws CanalClientException;
        void disconnect() throws CanalClientException;
    
        boolean checkValid() throws CanalClientException;
    
        void subscribe(String var1) throws CanalClientException;
        void subscribe() throws CanalClientException;
    
        void unsubscribe() throws CanalClientException;
    
        Message get(int var1) throws CanalClientException;
        Message get(int var1, Long var2, TimeUnit var3) throws CanalClientException;
    
        Message getWithoutAck(int var1) throws CanalClientException;
        Message getWithoutAck(int var1, Long var2, TimeUnit var3) throws CanalClientException;
    
        void ack(long var1) throws CanalClientException;
    
        void rollback(long var1) throws CanalClientException;
        void rollback() throws CanalClientException;
    }

**CanalConnectors：客户端连接建立类**

public class CanalConnectors {
        public CanalConnectors() {
        }
    
        public static CanalConnector newSingleConnector(SocketAddress address, String destination, String username, String password) {
            SimpleCanalConnector canalConnector = new SimpleCanalConnector(address, username, password, destination);
            canalConnector.setSoTimeout(60000);
            canalConnector.setIdleTimeout(3600000);
            return canalConnector;
        }
    
        public static CanalConnector newClusterConnector(List<? extends SocketAddress> addresses, String destination, String username, String password) {
            ClusterCanalConnector canalConnector = new ClusterCanalConnector(username, password, destination, new SimpleNodeAccessStrategy(addresses));
            canalConnector.setSoTimeout(60000);
            canalConnector.setIdleTimeout(3600000);
            return canalConnector;
        }
    
        public static CanalConnector newClusterConnector(String zkServers, String destination, String username, String password) {
            ClusterCanalConnector canalConnector = new ClusterCanalConnector(username, password, destination, new ClusterNodeAccessStrategy(destination, ZkClientx.getZkClient(zkServers)));
            canalConnector.setSoTimeout(60000);
            canalConnector.setIdleTimeout(3600000);
            return canalConnector;
        }
    }

**SimpleCanalConnector：客户端直连canal server**

public class SimpleCanalConnector implements CanalConnector {
        private static final Logger logger = LoggerFactory.getLogger(SimpleCanalConnector.class);
        private SocketAddress address;        //canal server地址
        private String username;              //用户名
        private String password;              //密码
        private int soTimeout;                //读取数据超时时间
        private int idleTimeout;              //空闲超时时间
        private String filter;                //订阅的数据库、表
        private final ByteBuffer readHeader;
        private final ByteBuffer writeHeader;
        private SocketChannel channel;
        private ReadableByteChannel readableChannel;
        private WritableByteChannel writableChannel;
        private List<Compression> supportedCompressions;
        private ClientIdentity clientIdentity;
        private ClientRunningMonitor runningMonitor;
        private ZkClientx zkClientx;
        private BooleanMutex mutex;
        private volatile boolean connected;
        private boolean rollbackOnConnect;
        private boolean rollbackOnDisConnect;
        private boolean lazyParseEntry;        //默认为false，返回的message为entry数组
                                               //若设置为true，则返回rawEntry
        private Object readDataLock;
        private Object writeDataLock;
        private volatile boolean running;
    
    
    *********
    构造函数
    
        public SimpleCanalConnector(SocketAddress address, String username, String password, String destination) {
        public SimpleCanalConnector(SocketAddress address, String username, String password, String destination, int soTimeout) {
        public SimpleCanalConnector(SocketAddress address, String username, String password, String destination, int soTimeout, int idleTimeout) {
    
    
    *********
    建立、断开连接
    
        public void connect() throws CanalClientException {
        public void disconnect() throws CanalClientException {
    
    
    *********
    订阅、取消订阅
    
        public void subscribe() throws CanalClientException {
            this.subscribe("");
        }
    
        public void subscribe(String filter) throws CanalClientException {
                    //客户端订阅的数据库、表
    
        public void unsubscribe() throws CanalClientException {
    
    
    *********
    get、ack、rollback
    
        public Message get(int batchSize) throws CanalClientException {
            return this.get(batchSize, (Long)null, (TimeUnit)null);
        }
    
        public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
            Message message = this.getWithoutAck(batchSize, timeout, unit);
            this.ack(message.getId());
            return message;
        }
    
        public Message getWithoutAck(int batchSize) throws CanalClientException {
    
        public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
            this.waitClientRunning();
            if (!this.running) {
                return null;
            } else {
                try {
                    int size = batchSize <= 0 ? 1000 : batchSize;
                    long time = timeout != null && timeout >= 0L ? timeout : -1L;
                    if (unit == null) {
                        unit = TimeUnit.MILLISECONDS;
                    }
    
                    this.writeWithHeader(Packet.newBuilder().setType(PacketType.GET).setBody(Get.newBuilder().setAutoAck(false).setDestination(this.clientIdentity.getDestination()).setClientId(String.valueOf(this.clientIdentity.getClientId())).setFetchSize(size).setTimeout(time).setUnit(unit.ordinal()).build().toByteString()).build().toByteArray());
                    return this.receiveMessages();
                } catch (IOException var7) {
                    throw new CanalClientException(var7);
                }
            }
        }
    
        private Message receiveMessages() throws IOException {
            byte[] data = this.readNextPacket();
            return CanalMessageDeserializer.deserializer(data, this.lazyParseEntry);
        }   //使用CanalMessageDeserializer反序列化data，获取message
    
    
        public void ack(long batchId) throws CanalClientException {
    
        public void rollback(long batchId) throws CanalClientException {
        public void rollback() throws CanalClientException {
            this.waitClientRunning();
            this.rollback(0L);
        }
    
    
    *********
    其他方法
    
        public String getUsername() {
        public String getPassword() {
        public int getSoTimeout() {
        public int getIdleTimeout() {
        public SocketAddress getAddress() {
        public SocketAddress getNextAddress() {
    
    
        public void setFilter(String filter) {
        public void setSoTimeout(int soTimeout) {
        public void setIdleTimeout(int idleTimeout) {
        public void setZkClientx(ZkClientx zkClientx) {
        public void setLazyParseEntry(boolean lazyParseEntry) {
        public void setRollbackOnConnect(boolean rollbackOnConnect) {
        public void setRollbackOnDisConnect(boolean rollbackOnDisConnect) {
    
    
        public void stopRunning() {
        public boolean checkValid() {
        public boolean isLazyParseEntry() {
    
    
        private InetSocketAddress doConnect() throws CanalClientException {
        private void doDisconnect() throws CanalClientException {
        private void quietlyClose(Channel channel) {
    
        private Message receiveMessages() throws IOException {
    
        private void writeWithHeader(byte[] body) throws IOException {
        private void writeWithHeader(WritableByteChannel channel, byte[] body) throws IOException {
    
        private byte[] readNextPacket() throws IOException {
        private byte[] readNextPacket(ReadableByteChannel channel) throws IOException {
        private void read(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
    
        private void waitClientRunning() {
        private synchronized void initClientRunningMonitor(ClientIdentity clientIdentity) {

**CanalMessageDeserializer：反序列化数据，转换为Message**

public class CanalMessageDeserializer {
        public CanalMessageDeserializer() {
        }
    
        public static Message deserializer(byte[] data) {
            return deserializer(data, false);
        }
    
        public static Message deserializer(byte[] data, boolean lazyParseEntry) {
            try {
                if (data == null) {
                    return null;
                } else {
                    Packet p = Packet.parseFrom(data);
                    switch(p.getType()) {
                    case MESSAGES:
                        if (!p.getCompression().equals(Compression.NONE) && !p.getCompression().equals(Compression.COMPRESSIONCOMPATIBLEPROTO2)) {
                            throw new CanalClientException("compression is not supported in this connector");
                        }
    
                        Messages messages = Messages.parseFrom(p.getBody());
                        Message result = new Message(messages.getBatchId());
                        if (lazyParseEntry) {  //lazyParseEntry为true，设置rawEntry
                            result.setRawEntries(messages.getMessagesList());
                            result.setRaw(true);
                        } else {
                            Iterator var5 = messages.getMessagesList().iterator();
    
                            while(var5.hasNext()) {
                                ByteString byteString = (ByteString)var5.next();
                                result.addEntry(Entry.parseFrom(byteString));
                                                //否则设置entry数组
                            }
    
                            result.setRaw(false);
                        }
    
                        return result;
                    case ACK:
                        Ack ack = Ack.parseFrom(p.getBody());
                        throw new CanalClientException("something goes wrong with reason: " + ack.getErrorMessage());
                    default:
                        throw new CanalClientException("unexpected packet type: " + p.getType());
                    }
                }
            } catch (Exception var7) {
                throw new CanalClientException("deserializer failed", var7);
            }
        }
    }

**ClusterCanalConnector：集群连接**

public class ClusterCanalConnector implements CanalConnector {
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
        private String username;               //用户名
        private String password;               //密码
        private int soTimeout = 60000;         //读取超时时间
        private int idleTimeout = 3600000;     //空闲超时时间
        private int retryTimes = 3;            //重试次数
        private int retryInterval = 5000;      //重试间隔
        private CanalNodeAccessStrategy accessStrategy;    //节点访问策略：canal server静态列表、zookeeper获取节点信息
        private SimpleCanalConnector currentConnector;     //获取节点信息后，调用SImpleCanalConnector连接
        private String destination;                        //访问的canal server下的instance
        private String filter;                             //订阅的数据库、表
    
    
    **********
    构造方法
    
        public ClusterCanalConnector(String username, String password, String destination, CanalNodeAccessStrategy accessStrategy) {
            this.username = username;
            this.password = password;
            this.destination = destination;
            this.accessStrategy = accessStrategy;
        }
    
    
    **********
    建立、断开连接
    
        public void connect() throws CanalClientException {
        public void disconnect() throws CanalClientException {
    
    
    **********
    订阅、取消订阅
    
        public void subscribe() throws CanalClientException {
            this.subscribe("");
        }
    
        public void subscribe(String filter) throws CanalClientException {
    
        public void unsubscribe() throws CanalClientException {
    
    
    **********
    get、ack、rollback
    
        public Message get(int batchSize) throws CanalClientException {
        public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
    
        public Message getWithoutAck(int batchSize) throws CanalClientException {
        public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
    
        public void rollback(long batchId) throws CanalClientException {
        public void rollback() throws CanalClientException {
    
        public void ack(long batchId) throws CanalClientException {
    
    
    **********
    其他方法
    
        public int getSoTimeout() {
        public int getIdleTimeout() {
        public String getUsername() {
        public String getPassword() {
        public int getRetryTimes() {
        public int getRetryInterval() {
        public CanalNodeAccessStrategy getAccessStrategy() {
        public SimpleCanalConnector getCurrentConnector() {
    
        public void setSoTimeout(int soTimeout) {
        public void setIdleTimeout(int idleTimeout) {
        public void setUsername(String username) {
        public void setPassword(String password) {
        public void setRetryTimes(int retryTimes) {
        public void setRetryInterval(int retryInterval) {
        public void setAccessStrategy(CanalNodeAccessStrategy accessStrategy) {
    
    
        public boolean checkValid() {
        public void stopRunning() {
    
        private void restart() throws CanalClientException {

**CanalNodeStrategy：节点访问策略**

public interface CanalNodeAccessStrategy {
    
        SocketAddress currentNode();
        SocketAddress nextNode();
    }

**SimpleNodeAccessStrategy：客户端轮询canal server服务列表建立连接**

public class SimpleNodeAccessStrategy implements CanalNodeAccessStrategy {
        private List<SocketAddress> nodes = new ArrayList();
        private int index = 0;
    
        public SimpleNodeAccessStrategy(List<? extends SocketAddress> nodes) {
            if (nodes != null && nodes.size() >= 1) {
                this.nodes.addAll(nodes);
            } else {
                throw new IllegalArgumentException("at least 1 node required.");
            }
        }
    
        public SocketAddress nextNode() {
            SocketAddress var1;
            try {
                var1 = (SocketAddress)this.nodes.get(this.index);
            } finally {
                this.index = (this.index + 1) % this.nodes.size();
            }
    
            return var1;
        }     //轮询列表，放回节点
    
        public SocketAddress currentNode() {
            return (SocketAddress)this.nodes.get(this.index);
        }
    }

**ClusterNodeAccessStrategy：从zookeeper获取canal server节点信息，建立连接**

public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
        private String destination;
        private IZkChildListener childListener;
        private IZkDataListener dataListener;
        private ZkClientx zkClient;
        private volatile List<InetSocketAddress> currentAddress = new ArrayList();
        private volatile InetSocketAddress runningAddress = null;
    
    
        public ClusterNodeAccessStrategy(String destination, ZkClientx zkClient) {
    
        public SocketAddress currentNode() {
            return this.nextNode();
        }
    
        public SocketAddress nextNode() {
            if (this.runningAddress != null) {
                return this.runningAddress;        //当前节点可用，返回当前节点
            } else if (!this.currentAddress.isEmpty()) {
                return (SocketAddress)this.currentAddress.get(0);
                                                  //返回节点列表中的第一个节点
            } else {
                throw new ServerNotFoundException("no alive canal server for " + this.destination);
            }
        }
    
        public void setZkClient(ZkClientx zkClient) {
        public ZkClientx getZkClient() {
    
        private void initClusters(List<String> currentChilds) {
        private void initRunning(Object data) {

**Message：客户端拉取的信息**

public class Message implements Serializable {
        private static final long serialVersionUID = 1234034768477580009L;
        private long id;
        private List<Entry> entries = new ArrayList();
        private boolean raw = true;
        private List<ByteString> rawEntries = new ArrayList();
    
    
    ********
    构造方法
    
        public Message(long id, List<Entry> entries) {
            this.id = id;
            this.entries = (List)(entries == null ? new ArrayList() : entries);
            this.raw = false;
        }
    
        public Message(long id, boolean raw, List entries) {
            this.id = id;
            if (raw) {
                this.rawEntries = (List)(entries == null ? new ArrayList() : entries);
            } else {
                this.entries = (List)(entries == null ? new ArrayList() : entries);
            }
    
            this.raw = raw;
        }
    
        public Message(long id) {
            this.id = id;
        }
    
    
    ********
    普通方法
    
        public long getId() {
        public void setId(long id) {
    
        public void setEntries(List<Entry> entries) {
        public void addEntry(Entry entry) {
        public List<Entry> getEntries() {
    
        public void setRawEntries(List<ByteString> rawEntries) {
        public void addRawEntry(ByteString rawEntry) {
        public List<ByteString> getRawEntries() {
    
        public boolean isRaw() {
        public void setRaw(boolean raw) {
    
        public String toString() {

**Entry**

public final class CanalEntry {
    
    **********
    内部类：CanalEntry.Entry
    
        public static final class Entry extends GeneratedMessage implements CanalEntry.EntryOrBuilder {
            private static final CanalEntry.Entry defaultInstance = new CanalEntry.Entry(true);
            private final UnknownFieldSet unknownFields;
    
            public static Parser<CanalEntry.Entry> PARSER = new AbstractParser<CanalEntry.Entry>() {
                public CanalEntry.Entry parsePartialFrom(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
                    return new CanalEntry.Entry(input, extensionRegistry);
                }
            };
    
            private int bitField0_;
            public static final int HEADER_FIELD_NUMBER = 1;
            private CanalEntry.Header header_;
            public static final int ENTRYTYPE_FIELD_NUMBER = 2;
            private CanalEntry.EntryType entryType_;
            public static final int STOREVALUE_FIELD_NUMBER = 3;
            private ByteString storeValue_;
            private byte memoizedIsInitialized;
            private int memoizedSerializedSize;
            private static final long serialVersionUID = 0L;
    
    *******
    构造方法
    
            private Entry(com.google.protobuf.GeneratedMessage.Builder<?> builder) {
            private Entry(boolean noInit) {
            private Entry(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
    
    *******
    常用方法
    
            public boolean hasHeader() {
            public CanalEntry.Header getHeader() {
            public CanalEntry.HeaderOrBuilder getHeaderOrBuilder() {
    
            public boolean hasEntryType() {
            public CanalEntry.EntryType getEntryType() {
    
            public boolean hasStoreValue() {
            public ByteString getStoreValue() {
    
    
    *******
    其他方法
    
            public static CanalEntry.Entry getDefaultInstance() {
            public static final Descriptor getDescriptor() {
    
            public Parser<CanalEntry.Entry> getParserForType() {
            public CanalEntry.Entry getDefaultInstanceForType() {
    
            public final UnknownFieldSet getUnknownFields() {
    
            protected FieldAccessorTable internalGetFieldAccessorTable() {
            protected Object writeReplace() throws ObjectStreamException {
            private void initFields() {
    
            public final boolean isInitialized() {
            public void writeTo(CodedOutputStream output) throws IOException {
    
            public int getSerializedSize() {
    
            public static CanalEntry.Entry parseFrom(ByteString data) throws InvalidProtocolBufferException {
            public static CanalEntry.Entry parseFrom(ByteString data, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
            public static CanalEntry.Entry parseFrom(byte[] data) throws InvalidProtocolBufferException {
            public static CanalEntry.Entry parseFrom(byte[] data, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
            public static CanalEntry.Entry parseFrom(InputStream input) throws IOException {
            public static CanalEntry.Entry parseFrom(InputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
            public static CanalEntry.Entry parseFrom(CodedInputStream input) throws IOException {
            public static CanalEntry.Entry parseFrom(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
            public static CanalEntry.Entry parseDelimitedFrom(InputStream input) throws IOException {
            public static CanalEntry.Entry parseDelimitedFrom(InputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
    
            public static CanalEntry.Entry.Builder newBuilder() {
            public static CanalEntry.Entry.Builder newBuilder(CanalEntry.Entry prototype) {
    
            public CanalEntry.Entry.Builder newBuilderForType() {
            public CanalEntry.Entry.Builder toBuilder() {
            protected CanalEntry.Entry.Builder newBuilderForType(BuilderParent parent) {

**EntryType：entry类型**

public static enum EntryType implements ProtocolMessageEnum {
            TRANSACTIONBEGIN(0, 1),       //事务开始
            ROWDATA(1, 2),                //数据内容
            TRANSACTIONEND(2, 3),         //事务结束
            HEARTBEAT(3, 4),              //心跳数据
            GTIDLOG(4, 5);                //gitd日志
    
            public static final int TRANSACTIONBEGIN_VALUE = 1;
            public static final int ROWDATA_VALUE = 2;
            public static final int TRANSACTIONEND_VALUE = 3;
            public static final int HEARTBEAT_VALUE = 4;
            public static final int GTIDLOG_VALUE = 5;
            private static EnumLiteMap<CanalEntry.EntryType> internalValueMap = new EnumLiteMap<CanalEntry.EntryType>() {
                public CanalEntry.EntryType findValueByNumber(int number) {
                    return CanalEntry.EntryType.valueOf(number);
                }
            };
            private static final CanalEntry.EntryType[] VALUES = values();
            private final int index;
            private final int value;

**Header**

public static final class Header extends GeneratedMessage implements CanalEntry.HeaderOrBuilder {
            private static final CanalEntry.Header defaultInstance = new CanalEntry.Header(true);
            private final UnknownFieldSet unknownFields;
            public static Parser<CanalEntry.Header> PARSER = new AbstractParser<CanalEntry.Header>() {
                public CanalEntry.Header parsePartialFrom(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
                    return new CanalEntry.Header(input, extensionRegistry);
                }
            };
            private int bitField0_;
            public static final int VERSION_FIELD_NUMBER = 1;
            private int version_;
            public static final int LOGFILENAME_FIELD_NUMBER = 2;
            private Object logfileName_;
            public static final int LOGFILEOFFSET_FIELD_NUMBER = 3;
            private long logfileOffset_;
            public static final int SERVERID_FIELD_NUMBER = 4;
            private long serverId_;
            public static final int SERVERENCODE_FIELD_NUMBER = 5;
            private Object serverenCode_;
            public static final int EXECUTETIME_FIELD_NUMBER = 6;
            private long executeTime_;
            public static final int SOURCETYPE_FIELD_NUMBER = 7;
            private CanalEntry.Type sourceType_;
            public static final int SCHEMANAME_FIELD_NUMBER = 8;
            private Object schemaName_;
            public static final int TABLENAME_FIELD_NUMBER = 9;
            private Object tableName_;
            public static final int EVENTLENGTH_FIELD_NUMBER = 10;
            private long eventLength_;
            public static final int EVENTTYPE_FIELD_NUMBER = 11;
            private CanalEntry.EventType eventType_;
            public static final int PROPS_FIELD_NUMBER = 12;
            private List<CanalEntry.Pair> props_;
            public static final int GTID_FIELD_NUMBER = 13;
            private Object gtid_;
            private byte memoizedIsInitialized;
            private int memoizedSerializedSize;
            private static final long serialVersionUID = 0L;
    
    *********
    构造方法
    
            private Header(com.google.protobuf.GeneratedMessage.Builder<?> builder) {
            private Header(boolean noInit) {
            private Header(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
    
    *********
    常用方法
    
            public boolean hasLogfileName() {            //二进制日志文件名称
            public String getLogfileName() {
            public ByteString getLogfileNameBytes() {
    
            public boolean hasLogfileOffset() {          //二进制偏移量
            public long getLogfileOffset() {
    
            public boolean hasSchemaName() {             //数据库名称
            public String getSchemaName() {
            public ByteString getSchemaNameBytes() {
    
            public boolean hasTableName() {               //表的名称
            public String getTableName() {
            public ByteString getTableNameBytes() {
    
            public boolean hasEventType() {               //事件类型，如：insert、delete等
            public CanalEntry.EventType getEventType() {
    
    
    *********
    其他方法
    
            public static CanalEntry.Header getDefaultInstance() {
            public CanalEntry.Header getDefaultInstanceForType() {
    
            public final UnknownFieldSet getUnknownFields() {
    
    
            public Parser<CanalEntry.Header> getParserForType() {
            public static final Descriptor getDescriptor() {
    
            protected FieldAccessorTable internalGetFieldAccessorTable() {
    
    
            public boolean hasVersion() {
            public int getVersion() {
    
            public boolean hasServerId() {
            public long getServerId() {
    
            public boolean hasServerenCode() {
            public String getServerenCode() {
            public ByteString getServerenCodeBytes() {
    
            public boolean hasExecuteTime() {
            public long getExecuteTime() {
    
            public boolean hasSourceType() {
            public CanalEntry.Type getSourceType() {
    
            public boolean hasEventLength() {
            public long getEventLength() {
    
            public int getPropsCount() {
            public CanalEntry.Pair getProps(int index) {
            public CanalEntry.PairOrBuilder getPropsOrBuilder(int index) {
            public List<CanalEntry.Pair> getPropsList() {
            public List<? extends CanalEntry.PairOrBuilder> getPropsOrBuilderList() {
    
            public boolean hasGtid() {
            public String getGtid() {
            public ByteString getGtidBytes() {
    
            private void initFields() {
    
            public final boolean isInitialized() {
    
    
            public void writeTo(CodedOutputStream output) throws IOException {
            public int getSerializedSize() {
    
            protected Object writeReplace() throws ObjectStreamException {
    
    
            public static CanalEntry.Header parseFrom(ByteString data) throws InvalidProtocolBufferException {
            public static CanalEntry.Header parseFrom(ByteString data, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
            public static CanalEntry.Header parseFrom(byte[] data) throws InvalidProtocolBufferException {
            public static CanalEntry.Header parseFrom(byte[] data, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
            public static CanalEntry.Header parseFrom(InputStream input) throws IOException {
            public static CanalEntry.Header parseFrom(InputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
            public static CanalEntry.Header parseDelimitedFrom(InputStream input) throws IOException {
            public static CanalEntry.Header parseDelimitedFrom(InputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
            public static CanalEntry.Header parseFrom(CodedInputStream input) throws IOException {
            public static CanalEntry.Header parseFrom(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
    
            public static CanalEntry.Header.Builder newBuilder() {
            public CanalEntry.Header.Builder newBuilderForType() {
            public static CanalEntry.Header.Builder newBuilder(CanalEntry.Header prototype) {
            public CanalEntry.Header.Builder toBuilder() {
    
            protected CanalEntry.Header.Builder newBuilderForType(BuilderParent parent) {

**RowChange**

public final class CanalEntry {
    
    ***********
    内部类：CanalEntry.RowChange
    
        public static final class RowChange extends GeneratedMessage implements CanalEntry.RowChangeOrBuilder {
            private static final CanalEntry.RowChange defaultInstance = new CanalEntry.RowChange(true);
            private final UnknownFieldSet unknownFields;
            public static Parser<CanalEntry.RowChange> PARSER = new AbstractParser<CanalEntry.RowChange>() {
                public CanalEntry.RowChange parsePartialFrom(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
                    return new CanalEntry.RowChange(input, extensionRegistry);
                }
            };
    
            private int bitField0_;
            public static final int TABLEID_FIELD_NUMBER = 1;
            private long tableId_;
            public static final int EVENTTYPE_FIELD_NUMBER = 2;
            private CanalEntry.EventType eventType_;
            public static final int ISDDL_FIELD_NUMBER = 10;
            private boolean isDdl_;
            public static final int SQL_FIELD_NUMBER = 11;
            private Object sql_;
            public static final int ROWDATAS_FIELD_NUMBER = 12;
            private List<CanalEntry.RowData> rowDatas_;
            public static final int PROPS_FIELD_NUMBER = 13;
            private List<CanalEntry.Pair> props_;
            public static final int DDLSCHEMANAME_FIELD_NUMBER = 14;
            private Object ddlSchemaName_;
            private byte memoizedIsInitialized;
            private int memoizedSerializedSize;
            private static final long serialVersionUID = 0L;
    
    ********
    构造方法
    
            private RowChange(com.google.protobuf.GeneratedMessage.Builder<?> builder) {
            private RowChange(boolean noInit) {
            private RowChange(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
    
    ********
    获取rowData
    
            public CanalEntry.RowData getRowDatas(int index) {
            public List<CanalEntry.RowData> getRowDatasList() {
    
            public int getRowDatasCount() {
    
    ********
    解析获取RowChange
    
            public static CanalEntry.RowChange parseFrom(ByteString data) throws InvalidProtocolBufferException {
            public static CanalEntry.RowChange parseFrom(ByteString data, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
            public static CanalEntry.RowChange parseFrom(byte[] data) throws InvalidProtocolBufferException {
            public static CanalEntry.RowChange parseFrom(byte[] data, ExtensionRegistryLite extensionRegistry) throws InvalidProtocolBufferException {
    
            public static CanalEntry.RowChange parseFrom(InputStream input) throws IOException {
            public static CanalEntry.RowChange parseFrom(InputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
            public static CanalEntry.RowChange parseDelimitedFrom(InputStream input) throws IOException {
            public static CanalEntry.RowChange parseDelimitedFrom(InputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
            public static CanalEntry.RowChange parseFrom(CodedInputStream input) throws IOException {
            public static CanalEntry.RowChange parseFrom(CodedInputStream input, ExtensionRegistryLite extensionRegistry) throws IOException {
    
    
    ********
    其他方法
    
            public boolean hasSql() {
            public boolean hasIsDdl() {
            public boolean hasTableId() {
            public boolean hasEventType() {
    
            public String getSql() {
            public long getTableId() {
            public boolean getIsDdl() {
            public CanalEntry.EventType getEventType() {
    
            public static CanalEntry.RowChange getDefaultInstance() {
            public static final Descriptor getDescriptor() {
    
            public final UnknownFieldSet getUnknownFields() {
            public CanalEntry.RowChange getDefaultInstanceForType() {
    
            protected FieldAccessorTable internalGetFieldAccessorTable() {
    
            public ByteString getSqlBytes() {
            public Parser<CanalEntry.RowChange> getParserForType() {
    
    
            public CanalEntry.RowDataOrBuilder getRowDatasOrBuilder(int index) {
            public List<? extends CanalEntry.RowDataOrBuilder> getRowDatasOrBuilderList() {
    
            public int getPropsCount() {
            public CanalEntry.Pair getProps(int index) {
            public List<CanalEntry.Pair> getPropsList() {
            public List<? extends CanalEntry.PairOrBuilder> getPropsOrBuilderList() {
            public CanalEntry.PairOrBuilder getPropsOrBuilder(int index) {
    
            public boolean hasDdlSchemaName() {
            public String getDdlSchemaName() {
            public ByteString getDdlSchemaNameBytes() {
    
            private void initFields() {
            public final boolean isInitialized() {
    
            public void writeTo(CodedOutputStream output) throws IOException {
    
            public int getSerializedSize() {
            protected Object writeReplace() throws ObjectStreamException {
    
    
            public static CanalEntry.RowChange.Builder newBuilder() {
            public CanalEntry.RowChange.Builder newBuilderForType() {
            public static CanalEntry.RowChange.Builder newBuilder(CanalEntry.RowChange prototype) {
            public CanalEntry.RowChange.Builder toBuilder() {
    
            protected CanalEntry.RowChange.Builder newBuilderForType(BuilderParent parent) {

**EventType：事件类型**

public static enum EventType implements ProtocolMessageEnum {
            INSERT(0, 1),
            UPDATE(1, 2),
            DELETE(2, 3),
            CREATE(3, 4),
            ALTER(4, 5),
            ERASE(5, 6),
            QUERY(6, 7),
            TRUNCATE(7, 8),
            RENAME(8, 9),
            CINDEX(9, 10),
            DINDEX(10, 11),
            GTID(11, 12),
            XACOMMIT(12, 13),
            XAROLLBACK(13, 14),
            MHEARTBEAT(14, 15);
    
            public static final int INSERT_VALUE = 1;
            public static final int UPDATE_VALUE = 2;
            public static final int DELETE_VALUE = 3;
            public static final int CREATE_VALUE = 4;
            public static final int ALTER_VALUE = 5;
            public static final int ERASE_VALUE = 6;
            public static final int QUERY_VALUE = 7;
            public static final int TRUNCATE_VALUE = 8;
            public static final int RENAME_VALUE = 9;
            public static final int CINDEX_VALUE = 10;
            public static final int DINDEX_VALUE = 11;
            public static final int GTID_VALUE = 12;
            public static final int XACOMMIT_VALUE = 13;
            public static final int XAROLLBACK_VALUE = 14;
            public static final int MHEARTBEAT_VALUE = 15;
            private static EnumLiteMap<CanalEntry.EventType> internalValueMap = new EnumLiteMap<CanalEntry.EventType>() {
                public CanalEntry.EventType findValueByNumber(int number) {
                    return CanalEntry.EventType.valueOf(number);
                }
            };
            private static final CanalEntry.EventType[] VALUES = values();
            private final int index;
            private final int value;