content: package com.alibaba.otter.canal.parse;
  
  import java.io.IOException;
  import java.io.UnsupportedEncodingException;
  import java.net.InetSocketAddress;
  import java.nio.charset.Charset;
  import java.util.BitSet;
  import java.util.List;
  import java.util.concurrent.ArrayBlockingQueue;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.atomic.AtomicLong;
  
  import org.junit.Ignore;
  
  import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector;
  import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
  import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
  import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
  import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
  import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
  import com.taobao.tddl.dbsync.binlog.LogBuffer;
  import com.taobao.tddl.dbsync.binlog.LogContext;
  import com.taobao.tddl.dbsync.binlog.LogDecoder;
  import com.taobao.tddl.dbsync.binlog.LogEvent;
  import com.taobao.tddl.dbsync.binlog.event.*;
  import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
  
  @Ignore
  public class MysqlBinlogParsePerformanceTest {
  
      protected static Charset charset = Charset.forName("utf-8");
  
      public static void main(String args[]) {
          try (DirectLogFetcher fetcher = new DirectLogFetcher()) {
              MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
              connector.connect();
              updateSettings(connector);
              sendBinlogDump(connector, "mysql-bin.000006", 120L, 3);
              fetcher.start(connector.getChannel());
              final BlockingQueue<LogBuffer> buffer = new ArrayBlockingQueue<>(1024);
              Thread thread = new Thread(() -> {
                  try {
                      consumer(buffer);
                  } catch (Throwable e) {
                      e.printStackTrace();
                  }
              });
              thread.start();
  
              while (fetcher.fetch()) {
                  buffer.put(fetcher.duplicate());
                  fetcher.consume(fetcher.limit());
              }
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  
      public static void consumer(BlockingQueue<LogBuffer> buffer) throws Throwable {
          LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
          LogContext context = new LogContext();
  
          AtomicLong sum = new AtomicLong(0);
          long start = System.currentTimeMillis();
          long last = 0;
          long end = 0;
          while (true) {
              LogEvent event = null;
              event = decoder.decode(buffer.take(), context);
              processEvent(event, decoder, context, sum);
              long current = sum.get();
              if (current - last >= 100000) {
                  end = System.currentTimeMillis();
                  long tps = ((current - last) * 1000) / (end - start);
                  System.out.println(" total : " + sum + " , cost : " + (end - start) + " , tps : " + tps);
                  last = current;
                  start = end;
              }
          }
      }
  
      private static void processEvent(LogEvent event, LogDecoder decoder, LogContext context, AtomicLong sum) throws Throwable {
          int eventType = event.getHeader().getType();
          switch (eventType) {
              case LogEvent.ROTATE_EVENT:
                  break;
              case LogEvent.WRITE_ROWS_EVENT_V1:
              case LogEvent.WRITE_ROWS_EVENT:
                  parseRowsEvent((WriteRowsLogEvent) event, sum);
                  break;
              case LogEvent.UPDATE_ROWS_EVENT_V1:
              case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
              case LogEvent.UPDATE_ROWS_EVENT:
                  parseRowsEvent((UpdateRowsLogEvent) event, sum);
                  break;
              case LogEvent.DELETE_ROWS_EVENT_V1:
              case LogEvent.DELETE_ROWS_EVENT:
                  parseRowsEvent((DeleteRowsLogEvent) event, sum);
                  break;
              case LogEvent.XID_EVENT:
                  sum.incrementAndGet();
                  break;
              case LogEvent.QUERY_EVENT:
                  sum.incrementAndGet();
                  break;
              case LogEvent.TRANSACTION_PAYLOAD_EVENT:
                  List<LogEvent> events = decoder.processIterateDecode(event, context);
                  for (LogEvent deEvent : events) {
                      processEvent(deEvent, decoder, context, sum);
                  }
                  break;
              default:
                  break;
          }
      }
  
      private static void sendBinlogDump(MysqlConnector connector, String binlogfilename, Long binlogPosition, int slaveId)
                                                                                                                           throws IOException {
          BinlogDumpCommandPacket binlogDumpCmd = new BinlogDumpCommandPacket();
          binlogDumpCmd.binlogFileName = binlogfilename;
          binlogDumpCmd.binlogPosition = binlogPosition;
          binlogDumpCmd.slaveServerId = slaveId;
          byte[] cmdBody = binlogDumpCmd.toBytes();
  
          HeaderPacket binlogDumpHeader = new HeaderPacket();
          binlogDumpHeader.setPacketBodyLength(cmdBody.length);
          binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
          PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
      }
  
      private static void updateSettings(MysqlConnector connector) throws IOException {
          update("set @master_binlog_checksum= '@@global.binlog_checksum'", connector);
          update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'", connector);
      }
  
      public static void update(String cmd, MysqlConnector connector) throws IOException {
          MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector);
          exector.update(cmd);
      }
  
      public static void parseRowsEvent(RowsLogEvent event, AtomicLong sum) {
          try {
              RowsLogBuffer buffer = event.getRowsBuf(charset);
              BitSet columns = event.getColumns();
              BitSet changeColumns = event.getChangeColumns();
              while (buffer.nextOneRow(columns)) {
                  int type = event.getHeader().getType();
                  if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
                      parseOneRow(event, buffer, columns, true);
                  } else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
                      parseOneRow(event, buffer, columns, false);
                  } else {
                      parseOneRow(event, buffer, columns, false);
                      if (!buffer.nextOneRow(changeColumns, true)) {
                          break;
                      }
                      parseOneRow(event, buffer, changeColumns, true);
                  }
  
                  sum.incrementAndGet();
              }
          } catch (Exception e) {
              throw new RuntimeException("parse row data failed.", e);
          }
      }
  
      public static void parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter)
                                                                                                            throws UnsupportedEncodingException {
          TableMapLogEvent map = event.getTable();
          if (map == null) {
              throw new RuntimeException("not found TableMap with tid=" + event.getTableId());
          }
  
          final int columnCnt = map.getColumnCnt();
          final ColumnInfo[] columnInfo = map.getColumnInfo();
          for (int i = 0; i < columnCnt; i++) {
              if (!cols.get(i)) {
                  continue;
              }
  
              ColumnInfo info = columnInfo[i];
              buffer.nextValue(null, i, info.type, info.meta);
              if (buffer.isNull()) {
              } else {
                  buffer.getValue();
              }
          }
      }
  }
  