바이너리 로그 API와 리플리케이션 리스너

전제조건

• 최신 리플리케이션 리스너를 복사합니다. MySQL Labs 또는 복제 소스로 부터 바이너리 파일을 다운로드 합니다.
• 최신 g++ (4.4 /4.5가 좋습니다.)와 Boost 1.45 또는 그 이상 버전
• OpenSSL
• Cmake
• Netbeans (선택사항)
• Clucence
• SOLR
• Linux (선택사항)
• MySQL 5.6
• 대담한 해커의 마음가짐!

기본 예제

/* example1.cpp */
#include 
#include "binlog_api.h"
int main(int argc, char** argv){
  if (argc != 2)  {
    fprintf(stderr,"Usage:\n\treplaybinlog URL\n\nExample:\n\treplaybinlog mysql://root:mypasswd@127.0.0.1:3306\n\n");
    return (EXIT_FAILURE);
  }
  Binary_log binlog(system::create_transport(argv[1]));
  if (binlog.connect())
  {
    fprintf(stderr,"Can't connect to the master.\n");
    return (EXIT_FAILURE);
  }
  if (binlog.set_position(4) != ERR_OK)
  {
    fprintf(stderr,"Can't reposition the binary log reader.\n");
    return (EXIT_FAILURE);
  }
return EXIT_SUCCESS;
}

% g++ example1.cpp -I/path/to/mysql-replication-listener/include/ -L/path/to/mysql-replication-listener/include/ -lreplication -lboost_system -o example1 

• 이벤트 루프.
• 콘텐츠 핸들러 파이프라인.

고급예제

#include <stdlib.h>
#include <boost/foreach.hpp>
#include "binlog_api.h"
#include "table_update.h"
#include "table_delete.h"
#include "table_insert.h"
#include "table_index.h"
using mysql::system::create_transport;
using mysql::Binary_log;
std::string cl_index_file;
class Incident_handler : public mysql::Content_handler
{
public:
 Incident_handler() : mysql::Content_handler() {}
 Binary_log_event *process_event(mysql::Incident_event *incident)
 {
   std::cout << "Event type: "
             << mysql::system::get_event_type_str(incident->get_event_type())
             << " length: " << incident->header()->event_length
             << " next pos: " << incident->header()->next_position
             << std::endl;
   std::cout << "type= "
             << (unsigned)incident->type
             << " message= "
             << incident->message
             <<  std::endl
             <<  std::endl;
   /* Consume the event */
   delete incident;
   return 0;
 }
};
class Applier : public mysql::Content_handler
{
public:
 Applier(Table_index *index)
 {
   m_table_index= index;
 }
 mysql::Binary_log_event *process_event(mysql::Row_event *rev)
 {
   boost::uint64_t table_id= rev->table_id;
   Int2event_map::iterator ti_it= m_table_index->find(table_id);
   if (ti_it == m_table_index->end ())
   {
     std::cout << "Table id "
               << table_id
               << " was not registered by any preceding table map event."
               << std::endl;
     return rev;
   }
   /*
    Each row event contains multiple rows and fields. The Row_iterator
    allows us to iterate one row at a time.
   */
   mysql::Row_event_set rows(rev, ti_it->second);
   /*
    Create a fuly qualified table name
   */
   std::ostringstream os;
   os << ti_it->second->db_name << '.' << ti_it->second->table_name;
   mysql::Row_event_set::iterator it= rows.begin();
   do {
     mysql::Row_of_fields fields= *it;
     if (rev->get_event_type() == mysql::WRITE_ROWS_EVENT)
       table_insert(os.str(),fields);
     if (rev->get_event_type() == mysql::UPDATE_ROWS_EVENT)
     {
       ++it;
       mysql::Row_of_fields fields2= *it;
       table_update(os.str(),fields,fields2);
     }
     if (rev->get_event_type() == mysql::DELETE_ROWS_EVENT)
       table_delete(os.str(),fields);
     } while (++it != rows.end());
     /* Consume the event */
     delete rev;
     return 0;
  }
private:
  Table_index *m_table_index;
};
int main(int argc, char** argv)
{
  if (argc != 3)
  {
    fprintf(stderr,"Usage:\n\nmysql2lucene URL\n\nExample:\n\nmysql2lucene mysql://root@127.0.0.1:3306 myindexfile\n\n");
    return (EXIT_FAILURE);
  }
  Binary_log binlog(create_transport(argv[1]));

  cl_index_file.append (argv[2]);
  /*
    Attach a custom event content handlers
  */
  Incident_handler incident_hndlr;
  Table_index table_event_hdlr;
  Applier replay_hndlr(&table_event_hdlr);
  binlog.content_handler_pipeline()->push_back(&table_event_hdlr);
  binlog.content_handler_pipeline()->push_back(&incident_hndlr);
  binlog.content_handler_pipeline()->push_back(&replay_hndlr);
  if (binlog.connect())
  {
    fprintf(stderr,"Can't connect to the master.\n");
    return (EXIT_FAILURE);
  }
  binlog.set_position("searchbin.000001", 4);
  bool quit= false;
  while(!quit)
  {
    /*
     Pull events from the master. This is the heart beat of the event listener.
    */
    Binary_log_event  *event;
    binlog.wait_for_next_event(&event);
    /*
     Print the event
    */
    std::cout << "Event type: "
              << mysql::system::get_event_type_str(event->get_event_type())
              << " length: " << event->header()->event_length
              << " next pos: " << event->header()->next_position
              << std::endl;
    /*
     Perform a special action based on event type
    */
    switch(event->header()->type_code)
    {
    case mysql::QUERY_EVENT:
      {
        const mysql::Query_event *qev= static_cast(event);
        std::cout << "query= "
                  << qev->query
                  << " db= "
                  << qev->db_name
                  <<  std::endl
                  <<  std::endl;
        if (qev->query.find("DROP TABLE REPLICATION_LISTENER") != std::string::npos)
        {
          quit= true;
        }
      }
      break;
    case mysql::ROTATE_EVENT:
      {
        mysql::Rotate_event *rot= static_cast(event);
        std::cout << "filename= "
                  << rot->binlog_file.c_str()
                  << " pos= "
                  << rot->binlog_pos
                  << std::endl
                  << std::endl;
      }
      break;
    } // end switch
    delete event;
  } // end loop
  return (EXIT_SUCCESS);
}

% ./mtr --start alias --mysqld=--log_bin=searchbin --mysqld=--binary_log=row

% java -jar start.jar

열 기반 이벤트 분석

• UPDATE_ROWS_EVENT
• DELETE_ROWS_EVENT
• INSERT_ROWS_EVENT

class Log_event_header
{
public:
uint8_t reserved;
uint32_t timestamp;
uint8_t type_code;
uint32_t server_id;
uint32_t event_length;
uint32_t next_position;
uint16_t flags;
};

class Row_event: public Binary_log_event
{
public:
uint64_t table_id;
uint16_t flags;
uint64_t columns_len;
uint32_t null_bits_len;
vector used_columns;
vector row;
[..]
};

if (row_event->header()->type_code= UPDATE_ROWS_EVENT)
{
[..]
  Row_event_set rows(row_event, table_map_event);
  Row_event_set::iterator it= rows.begin();

table map으로 부터의 인덱스 구축

class Table_map_handler : public Content_handler
{
  public:
  Binary_log_event *process_event(Table_map_event *event)
  {
    m_table_map.insert(Event_index_element(event->table_id,tm));
    /*
      Eat this event instead of passing it
      on to the next handler.
    */
    return NULL;
  }
  std::map m_table_map;
}
[..]
/* create an instance of the handler */
Table_map_handler table_index;
/* push content handler to stack */
binlog.content_handler_pipeline()->push_back(&table_index);
[..]

[..]
/* get table id from the recently captured row event */
boost::uint64_t table_id= row_event->table_id;
Table_map_event *table_map_event=
table_index.m_table_map[table_id];
[..]

반복의 마지막 행

Row_event_set rows(row_event, table_map_event);
do {
  Row_of_fields fields= *it;
  if (event->get_event_type() == UPDATE_ROWS_EVENT)
  {
    ++it;
    Row_of_fields fields2= *it;
    table_update(os.str(),fields,fields2);
  }
[..]
} while (++it != rows.end());

void table_update (const string& table_name,
Row_of_fields &old_fields,
Row_of_fields &new_fields
{
  int field_id= 0, string str;
  Row_of_fields::iterator
  field_it= new_fields.begin();
  Converter converter;
  do {
    converter.to(str, *field_it);
    cout << field_id << "= " << str;"
    ++field_it; ++field_id;
  } while(field_it != new_fields.end());
}

관련 링크