MySQLデータベース監視binlogを有効にする手順

序文

多くの場合、ユーザーが自分のデータに対して実行する操作に基づいて何かを行う必要があります。

たとえば、ユーザーがアカウントを削除した場合、私たちはそのユーザーを叱責し、再度アカウントに戻ってきてもらうよう懇願するテキストメッセージを送信します。

同様の機能はビジネスロジック層に実装でき、ユーザーの削除要求を受け取った後にこの操作を実行できますが、データベース binlog は別の操作方法を提供します。

binlog を監視するには、2 つの手順が必要です。最初の手順は、もちろん MySQL でこの機能を有効にすることです。2 番目の手順は、ログを読み取るプログラムを作成することです。

mysql は binlog をオンにします。

まず、MySQL の binlog は通常は開かれていないため、次のものが必要です。

mysql 構成ファイル my.cnf を見つけます。場所はオペレーティングシステムによって異なる場合があります。自分で見つけることができます。

これに次の内容を追加します。

[mysqld]
サーバーID = 1
ログ bin = mysql bin
binlog 形式 = ROW

次に、mysql を再起動します。

/ウブントゥ
サービスmysqlの再起動
// マック
mysql.server の再起動

正常に有効化されているかどうかを監視する

mysql コマンドラインを入力して実行します:

'%log_bin%' のような変数を表示します。

結果が以下の通りであれば成功です。

書き込まれているバイナリログのステータスを表示します。

binlog を読み取るコード

依存関係の導入

私たちはオープンソースの実装をいくつか使用しています。奇妙な理由から、mysql-binlog-connector-java パッケージ (公式 github リポジトリ) [github.com/shyiko/mysq…] を選択しました。具体的な依存関係は次のとおりです。

<!-- https://mvnrepository.com/artifact/com.github.shyiko/mysql-binlog-connector-java -->
 <依存関係>
 <groupId>com.github.shyiko</groupId>
 <artifactId>mysql-binlog-connector-java</artifactId>
 <バージョン>0.17.0</バージョン>
 </依存関係>

もちろん、binlog 処理用のオープンソース実装は多数あり、Alibaba の cancl もその 1 つであり、これを使用することもできます。

デモを書く

公式リポジトリの readme に従って、簡単にデモを書いてみましょう。

 パブリック静的voidメイン(String[] args) {
 BinaryLogClient クライアント = new BinaryLogClient("ホスト名", 3306, "ユーザー名", "パスワード");
 イベントデシリアライザーeventDeserializer = new EventDeserializer();
 イベントデシリアライザー.setCompatibilityMode(
 イベントデシリアライザー.互換性モード.DATE_AND_TIME_AS_LONG、
 イベントデシリアライザー.互換性モード.CHAR_AND_BINARY_AS_BYTE_ARRAY
 );
 クライアントにイベントデシリアライザーを設定します。
 クライアント.registerEventListener(新しいBinaryLogClient.EventListener() {

 @オーバーライド
 パブリック void onEvent(イベント イベント) {
 //やるべきこと
 何かを行う();
 ロガー情報(イベント.toString());
 }
 });
 クライアントに接続します。
 }

これは、公式チュートリアルに完全に従って記述されています。onEvent に独自のビジネスロジックを記述できます。私はテスト中なので、すべてのイベントを出力します。

その後、MySQL に手動でログインし、それぞれ追加、変更、削除の操作を実行しました。監視されたログは次のとおりです。

00:23:13.331 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=0、イベントタイプ=ROTATE、サーバーId=1、ヘッダー長=19、データ長=28、次の位置=0、フラグ=32}、データ=RotateEventData{binlogFilename='mysql-bin.000001'、binlogPosition=886}}
00:23:13.334 [main] INFO util.MysqlBinLog - イベント{header=EventHeaderV4{timestamp=1556468403000、eventType=FORMAT_DESCRIPTION、serverId=1、headerLength=19、dataLength=100、nextPosition=0、flags=0}、data=FormatDescriptionEventData{binlogVersion=4、serverVersion='5.7.23-0ubuntu0.16.04.1-log'、headerLength=19、dataLength=95}}
00:23:23.715 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468603000、イベントタイプ=ANONYMOUS_GTID、サーバーID=1、ヘッダー長=19、データ長=46、次の位置=951、フラグ=0}、データ=null}
00:23:23.716 [main] INFO util.MysqlBinLog - イベント{header=EventHeaderV4{timestamp=1556468603000、eventType=QUERY、serverId=1、headerLength=19、dataLength=51、nextPosition=1021、flags=8}、data=QueryEventData{threadId=4、executionTime=0、errorCode=0、database='pf'、sql='BEGIN'}}
00:23:23.721 [main] INFO util.MysqlBinLog - イベント{header=EventHeaderV4{timestamp=1556468603000、eventType=TABLE_MAP、serverId=1、headerLength=19、dataLength=32、nextPosition=1072、flags=0}、data=TableMapEventData{tableId=108、database='pf'、table='student'、columnTypes=15、3、columnMetadata=135、0、columnNullability={}}}
00:23:23.724 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468603000、イベントタイプ=EXT_WRITE_ROWS、サーバーId=1、ヘッダー長=19、データ長=23、次の位置=1114、フラグ=0}、データ=WriteRowsEventData{テーブルId=108、含まれる列={0、1}、行=[
[[B@546a03af, 2]
]}}
00:23:23.725 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468603000、イベントタイプ=XID、サーバーId=1、ヘッダー長=19、データ長=12、次の位置=1145、フラグ=0}、データ=XidEventData{xid=28}}
00:23:55.872 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468635000、イベントタイプ=ANONYMOUS_GTID、サーバーID=1、ヘッダー長=19、データ長=46、次の位置=1210、フラグ=0}、データ=null}
00:23:55.872 [main] INFO util.MysqlBinLog - イベント{header=EventHeaderV4{timestamp=1556468635000、eventType=QUERY、serverId=1、headerLength=19、dataLength=51、nextPosition=1280、flags=8}、data=QueryEventData{threadId=4、executionTime=0、errorCode=0、database='pf'、sql='BEGIN'}}
00:23:55.873 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468635000、イベントタイプ=TABLE_MAP、サーバーId=1、ヘッダー長=19、データ長=32、次の位置=1331、フラグ=0}、データ=TableMapEventData{テーブルId=108、データベース='pf'、テーブル='student'、列タイプ=15、3、列メタデータ=135、0、列Nullability={}}}
00:23:55.875 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468635000、イベントタイプ=EXT_UPDATE_ROWS、サーバーId=1、ヘッダー長=19、データ長=31、次の位置=1381、フラグ=0}、データ=UpdateRowsEventData{テーブルId=108、含まれている列のBeforeUpdate={0, 1}、含まれている列={0, 1}、行=[
{前=[[B@6833ce2c, 1]、後=[[B@725bef66, 3]}
]}}
00:23:55.875 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468635000、イベントタイプ=XID、サーバーId=1、ヘッダー長=19、データ長=12、次の位置=1412、フラグ=0}、データ=XidEventData{xid=41}}
00:24:22.333 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468662000、イベントタイプ=ANONYMOUS_GTID、サーバーID=1、ヘッダー長=19、データ長=46、次の位置=1477、フラグ=0}、データ=null}
00:24:22.334 [main] INFO util.MysqlBinLog - イベント{header=EventHeaderV4{timestamp=1556468662000、eventType=QUERY、serverId=1、headerLength=19、dataLength=51、nextPosition=1547、flags=8}、data=QueryEventData{threadId=4、executionTime=0、errorCode=0、database='pf'、sql='BEGIN'}}
00:24:22.334 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468662000、イベントタイプ=TABLE_MAP、サーバーId=1、ヘッダー長=19、データ長=32、次の位置=1598、フラグ=0}、データ=TableMapEventData{テーブルId=108、データベース='pf'、テーブル='student'、列タイプ=15、3、列メタデータ=135、0、列Nullability={}}}
00:24:22.335 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468662000、イベントタイプ=EXT_DELETE_ROWS、サーバーId=1、ヘッダー長=19、データ長=23、次の位置=1640、フラグ=0}、データ=DeleteRowsEventData{テーブルId=108、含まれる列={0、1}、行=[
[[B@1888ff2c, 3]
]}}
00:24:22.335 [main] INFO util.MysqlBinLog - イベント{ヘッダー=EventHeaderV4{タイムスタンプ=1556468662000、イベントタイプ=XID、サーバーId=1、ヘッダー長=19、データ長=12、次の位置=1671、フラグ=0}、データ=XidEventData{xid=42}}

独自のビジネスに合わせて、より優れたカスタマイズされたツールクラスをカプセル化します

最初はコードを投稿するつもりでしたが、、、コードがどんどん長くなっていったのでGitHubにアップロードすることにしました。ここでは実装の一部だけ投稿します。コード転送ポータル

実装のアイデア

すべてのデータベース内のすべてのデータテーブルを監視する必要はないため、単一のテーブルの監視をサポートします。
複数のスレッドで使用できます。
監視対象のコンテンツを好みの形式に変換します (記事のデータ構造はあまり良くないかもしれませんが、これより適切なものは思いつきません)。

したがって、実装のアイデアはおおよそ次のようになります。

クライアントをカプセル化し、外部には取得メソッドのみを提供し、初期化の詳細コードは隠蔽します。
リスナー (疑似) を登録するメソッドを提供します。これにより、テーブルのリスナーを登録できます (リスナーインターフェイスを再定義し、登録されているすべてのリスナーはこれを実装するだけで済みます)。
唯一の実際のリスナーはクライアントであり、クライアントはこのデータベースインスタンス上のすべての操作をリッスンし、それらを目的の形式の LogItem に変換して、ブロッキングキューに配置します。
複数のスレッドを開始し、ブロッキングキューを消費し、LogItem に対応するデータテーブルリスナーを呼び出して、ビジネスロジックを実行します。

初期化コード:

 パブリックMysqlBinLogListener(Conf conf) {
 BinaryLogClient クライアント = new BinaryLogClient(conf.host、conf.port、conf.username、conf.passwd);
 イベントデシリアライザーeventDeserializer = new EventDeserializer();
 イベントデシリアライザー.setCompatibilityMode(
 イベントデシリアライザー.互換性モード.DATE_AND_TIME_AS_LONG、
 イベントデシリアライザー.互換性モード.CHAR_AND_BINARY_AS_BYTE_ARRAY
 );
 クライアントにイベントデシリアライザーを設定します。
 this.parseClient = クライアント;
 this.queue = 新しいArrayBlockingQueue<>(1024);
 conf を次のように変更します。
 リスナー = 新しい ConcurrentHashMap<>();
 dbTableCols = 新しいConcurrentHashMap<>();
 this.consumer = Executors.newFixedThreadPool(consumerThreads);
 }

登録コード:

 パブリック void regListener(String db、String table、BinLogListener listener) 例外をスローします {
 文字列 dbTable = getdbTable(db, table);
 クラス.forName("com.mysql.jdbc.Driver");
 // 現在登録されているテーブルの列情報を保存します。Connection connection = DriverManager.getConnection("jdbc:mysql://" + conf.host + ":" + conf.port, conf.username, conf.passwd);
 Map<String, Colum> cols = getColMap(connection, db, table);
 dbTableCols.put(dbTable、列);

 //現在登録されているリスナーを保存する
 リスト<BinLogListener> リスト = listeners.getOrDefault(dbTable, 新しい ArrayList<>());
 リストに追加します(リスナー)。
 listeners.put(dbTable、リスト);
 }

このステップでは、リスナーを登録しながらテーブルのスキーマ情報を取得し、それをマップに保存して、後続のデータ処理を容易にします。

リスニングコード:

 @オーバーライド
 パブリック void onEvent(イベント イベント) {
 イベントタイプ eventType = event.getHeader().getEventType();

 イベントタイプ == イベントタイプ.TABLE_MAP) {
 テーブルマップイベントデータ tableData = event.getData();
 文字列 db = tableData.getDatabase();
 文字列テーブル = tableData.getTable();
 dbTable = getdbTable(db、テーブル);
 }

 // 追加、削除、更新の3つの操作のみを処理します if (isWrite(eventType) || isUpdate(eventType) || isDelete(eventType)) {
 if (isWrite(イベントタイプ)) {
 WriteRowsEventData データ = event.getData();
 (Serializable[] 行: data.getRows()) の場合 {
  dbTableCols.containsKey(dbTable) の場合 {
  LogItem e = LogItem.itemFromInsert(行、dbTableCols.get(dbTable));
  e.setDbTable(dbTable);
  キューに追加します。
  }
 }
 }
 }
 }

面倒なので、ここでは追加操作の処理のみを実装し、他の操作は書いていません。

消費コード:

 パブリック void parse() は IOException をスローします {
 parseClient.registerEventListener(これを);

 (int i = 0; i < consumerThreads; i++) の場合 {
 消費者.送信(() -> {
 （真）の間{
  キューサイズ() > 0 の場合
  試す {
  ログアイテム項目 = queue.take();
  文字列 dbtable = item.getDbTable();
  リスナー.get(dbtable).forEach(l -> {
  l.onEvent(アイテム);
  });

  } キャッチ (InterruptedException e) {
  e.printStackTrace();
  }
  }
  スレッドをスリープ状態にします(1000);
 }
 });
 }
 parseClient.connect();
 }

消費するときは、キューからアイテムを取得し、対応する 1 つ以上のリスナーを取得してそれぞれアイテムを消費します。

テストコード:

 パブリック静的void main(String[] args)は例外をスローします{
 Conf conf = 新しい Conf();
 conf.host = "ホスト名";
 conf.port = 3306;
 conf.username = conf.passwd = "hhsgsb";

 MysqlBinLogListener を新しい MysqlBinLogListener(conf);
 mysqlBinLogListener.parseArgsAndRun(引数);
 mysqlBinLogListener.regListener("pf", "学生", 項目 -> {
 System.out.println(新しい文字列((byte[])item.getAfter().get("name")));
 logger.info("{} に挿入、値 = {}", item.getDbTable(), item.getAfter());
 });
 mysqlBinLogListener.regListener("pf", "teacher", item -> System.out.println("teacher ===="));

 mysqlBinLogListener.parse();
 }

この短いコードでは、2 つのリスナーが登録されており、それぞれ学生テーブルと教師テーブルをリッスンし、それぞれを印刷します。テスト後、教師テーブルにデータを挿入するときに、定義されたビジネスロジックを個別に実行できます。

注: ここでのツールクラスは、多くの例外処理が行われていないため直接使用できません。関数は挿入ステートメントのみをリッスンしますが、これは実装の参照として使用できます。

参考文献