日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關咨詢
選擇下列產品馬上在線溝通
服務時間：8:30-17:00
你可能遇到了下面的問題
關閉右側工具欄

新聞中心

這里有您想知道的互聯網營銷解決方案
Java代碼中，如何監(jiān)控MySQL的Binlog？

最近在工作中，遇到了這樣一個業(yè)務場景，我們需要關注一個業(yè)務系統數據庫中某幾張表的數據，當數據發(fā)生新增或修改時，將它同步到另一個業(yè)務系統數據庫中的表中。

創(chuàng)新互聯公司是一家專注于網站制作、成都網站建設與策劃設計,槐蔭網站建設哪家好?創(chuàng)新互聯公司做網站,專注于網站建設10年,網設計領域的專業(yè)建站公司;建站業(yè)務涵蓋:槐蔭等地區(qū)?；笔a做網站價格咨詢:028-86922220

一提到數據庫的同步，估計大家第一時間想到的就是基于binlog的主從復制了，但是放在我們的場景中，還有幾個問題：

  • 第一，并不是需要復制所有表的數據，復制對象只有少量的幾張表
  • 第二，也是比較麻煩的，兩個業(yè)務系統數據庫表結構可能不一致。例如，要同步數據庫1的A表中的某些字段到數據庫2的B表中，在這一過程中，A表和B表的字段并不是完全相同

這樣的話，我們只能通過代碼的方式，首先獲取到數據庫1表中數據的變動，再通過手動映射的方式，插入到數據庫2的表中。但是，獲取變動數據的這一過程，還是離不開binlog，因此我們就需要在代碼中對binlog進行一下監(jiān)控。

先說結論，我們最終使用了一個開源工具mysql-binlog-connector-java，用來監(jiān)控binlog變化并獲取數據，獲取數據后再手動插入到另一個庫的表中，基于它來實現了數據的同步。這個工具的git項目地址如下：

  
 
 
 
    
   
  
  
  
  1. https://github.com/shyiko/mysql-binlog-connector-java 
    2.

在正式開始前，還是先簡單介紹一下mysql的binlog，binlog是一個二進制文件，它保存在磁盤中，是用來記錄數據庫表結構變更、表數據修改的二進制日志。其實除了數據復制外，它還可以實現數據恢復、增量備份等功能。

啟動項目前，首先需要確保mysql服務已經啟用了binlog：

  
 
 
 
    
   
  
  
  
  1. show variables like 'log_bin'; 
    2.

如果為值為OFF，表示沒有啟用，那么需要首先啟用binlog，修改配置文件：

  
 
 
 
    
   
  
  
  
  1. log_bin=mysql-bin 
    2.    
  
  
  
  2. binlog-format=ROW 
    3.    
  
  
  
  3. server-id=1 
    4.

對參數做一個簡要說明：

  • 在配置文件中加入了log_bin配置項后，表示啟用了binlog
  • binlog-format是binlog的日志格式，支持三種類型，分別是STATEMENT、ROW、MIXED，我們在這里使用ROW模式
  • server-id用于標識一個sql語句是從哪一個server寫入的，這里一定要進行設置，否則我們在后面的代碼中會無法正常監(jiān)聽到事件

在更改完配置文件后，重啟mysql服務。再次查看是否啟用binlog，返回為ON，表示已經開啟成功。

在Java項目中，首先引入maven坐標：

  
 
 
 
    
   
  
  
  
  1.  
    2.    
  
  
  
  2.     com.github.shyiko 
    3.    
  
  
  
  3.     mysql-binlog-connector-java 
    4.    
  
  
  
  4.     0.21.0 
    5.    
  
  
  
  5.  
    6.

寫一段簡單的示例，看看它的具體使用方式：

  
 
 
 
    
   
  
  
  
  1. public static void main(String[] args) { 
    2.    
  
  
  
  2.     BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306, "hydra", "123456"); 
    3.    
  
  
  
  3.     client.setServerId(2); 
    4.    
  
  
  
  4.  
    5.    
  
  
  
  5.     client.registerEventListener(event -> { 
    6.    
  
  
  
  6.         EventData data = event.getData(); 
    7.    
  
  
  
  7.         if (data instanceof TableMapEventData) { 
    8.    
  
  
  
  8.             System.out.println("Table:"); 
    9.    
  
  
  
  9.             TableMapEventData tableMapEventData = (TableMapEventData) data; 
    10.    
  
  
  
  10.             System.out.println(tableMapEventData.getTableId()+": ["+tableMapEventData.getDatabase() + "-" + tableMapEventData.getTable()+"]"); 
    11.    
  
  
  
  11.         } 
    12.    
  
  
  
  12.         if (data instanceof UpdateRowsEventData) { 
    13.    
  
  
  
  13.             System.out.println("Update:"); 
    14.    
  
  
  
  14.             System.out.println(data.toString()); 
    15.    
  
  
  
  15.         } else if (data instanceof WriteRowsEventData) { 
    16.    
  
  
  
  16.             System.out.println("Insert:"); 
    17.    
  
  
  
  17.             System.out.println(data.toString()); 
    18.    
  
  
  
  18.         } else if (data instanceof DeleteRowsEventData) { 
    19.    
  
  
  
  19.             System.out.println("Delete:"); 
    20.    
  
  
  
  20.             System.out.println(data.toString()); 
    21.    
  
  
  
  21.         } 
    22.    
  
  
  
  22.     }); 
    23.    
  
  
  
  23.  
    24.    
  
  
  
  24.     try { 
    25.    
  
  
  
  25.         client.connect(); 
    26.    
  
  
  
  26.     } catch (IOException e) { 
    27.    
  
  
  
  27.         e.printStackTrace(); 
    28.    
  
  
  
  28.     } 
    29.    
  
  
  
  29. } 
    30.

首先，創(chuàng)建一個BinaryLogClient客戶端對象，初始化時需要傳入mysql的連接信息，創(chuàng)建完成后，給客戶端注冊一個監(jiān)聽器，來實現它對binlog的監(jiān)聽和解析。在監(jiān)聽器中，我們暫時只對4種類型的事件數據進行了處理，除了WriteRowsEventData、DeleteRowsEventData、UpdateRowsEventData對應增刪改操作類型的事件數據外，還有一個TableMapEventData類型的數據，包含了表的對應關系，在后面的例子中再具體說明。

在這里，客戶端監(jiān)聽到的是數據庫級別的所有事件，并且可以監(jiān)聽到表的DML語句和DDL語句，所以我們只需要處理我們關心的事件數據就行，否則會收到大量的冗余數據。

啟動程序，控制臺輸出：

  
 
 
 
    
   
  
  
  
  1. com.github.shyiko.mysql.binlog.BinaryLogClient openChannelToBinaryLogStream 
    2.    
  
  
  
  2. 信息: Connected to 127.0.0.1:3306 at mysql-bin.000002/1046 (sid:2, cid:10) 
    3.

信息: Connected to 127.0.0.1:3306 at mysql-bin.000002/1046 (sid:2, cid:10)

連接mysql的binlog成功，接下來，我們在數據庫中插入一條數據，這里操作的數據庫名字是tenant，表是dept：

  
 
 
 
    
   
  
  
  
  1. insert into dept VALUES(8,"人力","","1"); 
    2.

這時，控制臺就會打印監(jiān)聽到事件的數據：

  
 
 
 
    
   
  
  
  
  1. Table: 
    2.    
  
  
  
  2. 108: [tenant-dept] 
    3.    
  
  
  
  3. Insert: 
    4.    
  
  
  
  4. WriteRowsEventData{tableId=108, includedColumns={0, 1, 2, 3}, rows=[ 
    5.    
  
  
  
  5.     [8, 人力, , 1] 
    6.    
  
  
  
  6. ]} 
    7.

我們監(jiān)聽到的事件類型數據有兩類，第一類是TableMapEventData，通過它可以獲取操作的數據庫名稱、表名稱以及表的id。之所以我們要監(jiān)聽這個事件，是因為之后監(jiān)聽的實際操作中返回數據中包含了表的id，而沒有表名等信息，所以如果我們想知道具體的操作是在哪一張表的話，就要先維護一個id與表的對應關系。

第二個打印出來的監(jiān)聽事件數據是WriteRowsEventData，其中記錄了insert語句作用的表，插入涉及到的列，以及實際插入的數據。另外，如果我們只需要對特定的一張或幾張表進行處理的話，也可以提前設置表的名單，在這里根據表id到表名的映射關系，實現數據的過濾，

接下來，我們再執(zhí)行一條update語句：

  
 
 
 
    
   
  
  
  
  1. update dept set tenant_id=3 where id=8 or id=9 
    2.

控制臺輸出：

  
 
 
 
    
   
  
  
  
  1. Table: 
    2.    
  
  
  
  2. 108: [tenant-dept] 
    3.    
  
  
  
  3. Update: 
    4.    
  
  
  
  4. UpdateRowsEventData{tableId=108, includedColumnsBeforeUpdate={0, 1, 2, 3}, includedColumns={0, 1, 2, 3}, rows=[ 
    5.    
  
  
  
  5.     {before=[8, 人力, , 1], after=[8, 人力, , 3]}, 
    6.    
  
  
  
  6.     {before=[9, 人力, , 1], after=[9, 人力, , 3]} 
    7.    
  
  
  
  7. ]} 
    8.

在執(zhí)行update語句時，可能會作用于多條數據，因此在實際修改的數據中，可能包含多行記錄，這一點體現在上面的rows中，包含了id為8和9的兩條數據。

最后，再執(zhí)行一條delete語句：

  
 
 
 
    
   
  
  
  
  1. delete from dept where tenant_id=3 
    2.

控制臺打印如下，rows中同樣返回了生效的兩條數據：

  
 
 
 
    
   
  
  
  
  1. Table: 
    2.    
  
  
  
  2. 108: [tenant-dept] 
    3.    
  
  
  
  3. Delete: 
    4.    
  
  
  
  4. DeleteRowsEventData{tableId=108, includedColumns={0, 1, 2, 3}, rows=[ 
    5.    
  
  
  
  5.     [8, 人力, , 3], 
    6.    
  
  
  
  6.     [9, 人力, , 3] 
    7.    
  
  
  
  7. ]} 
    8.

簡單的使用原理介紹完成后，再回到我們原先的需求上，需要將一張表中新增或修改的數據同步到另一張表中，問題還有一個，就是如何將返回的數據對應到所在的列上。這時應該怎么實現呢?

以update操作為例，我們要對提取的數據后進行一下處理，更改上面例子中的方法：

  
 
 
 
    
   
  
  
  
  1. if (data instanceof UpdateRowsEventData) { 
    2.    
  
  
  
  2.     System.out.println("Update:"); 
    3.    
  
  
  
  3.     UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data; 
    4.    
  
  
  
  4.     for (Map.Entry row : updateRowsEventData.getRows()) { 
    5.    
  
  
  
  5.         List entries = Arrays.asList(row.getValue()); 
    6.    
  
  
  
  6.         System.out.println(entries); 
    7.    
  
  
  
  7.         JSONObject dataObject = getDataObject(entries); 
    8.    
  
  
  
  8.         System.out.println(dataObject); 
    9.    
  
  
  
  9.     } 
    10.    
  
  
  
  10. } 
    11.

在將data強制轉換為UpdateRowsEventData后，可以使用getRows方法獲取到更新的行數據，并且能夠取到每一列的值。

之后，調用了一個自己實現的getDataObject方法，用它來實現數據到列的綁定過程：

  
 
 
 
    
   
  
  
  
  1. private static JSONObject getDataObject(List message) { 
    2.    
  
  
  
  2.     JSONObject resultObject = new JSONObject(); 
    3.    
  
  
  
  3.     String format = "{\"id\":\"0\",\"dept_name\":\"1\",\"comment\":\"2\",\"tenant_id\":\"3\"}"; 
    4.    
  
  
  
  4.     JSONObject json = JSON.parseObject(format); 
    5.    
  
  
  
  5.     for (String key : json.keySet()) { 
    6.    
  
  
  
  6.         resultObject.put(key, message.get(json.getInteger(key))); 
    7.    
  
  
  
  7.     } 
    8.    
  
  
  
  8.     return resultObject; 
    9.    
  
  
  
  9. } 
    10.

在format字符串中，提前維護了一個數據庫表的字段順序的字符串，標識了每個字段位于順序中的第幾個位置。通過上面這個函數，能夠實現數據到列的填裝過程，我們再執(zhí)行一條update語句來查看一下結果：

  
 
 
 
    
   
  
  
  
  1. update dept set tenant_id=3,comment="1" where id=8 
    2.

控制臺打印結果如下：

  
 
 
 
    
   
  
  
  
  1. Table: 
    2.    
  
  
  
  2. 108: [tenant-dept] 
    3.    
  
  
  
  3. Update: 
    4.    
  
  
  
  4. [8, 人力, 1, 3] 
    5.    
  
  
  
  5. {"tenant_id":3,"dept_name":"人力","comment":"1","id":8} 
    6.

可以看到，將修改后的這一條記錄中的屬性填裝到了它對應的列中，之后我們再根據具體的業(yè)務邏輯，就可以根據字段名取出數據，將數據同步到其他的表了。

本文轉載自微信公眾號「碼農參上」，可以通過以下二維碼關注。轉載本文請聯系碼農參上公眾號。


當前標題：Java代碼中，如何監(jiān)控MySQL的Binlog？
URL地址：http://www.5511xx.com/article/dhsssji.html