萬字分享,我是如何一步一步監控公司MySQL的?
整理了一些Java方面的架構、面試資料(微服務、叢集、分散式、中介軟體等),有需要的小夥伴可以關注公眾號【程式設計師內點事】,無套路自行領取
更多優選
一口氣說出 9種 分散式ID生成方式,面試官有點懵了
- 3萬字總結,Mysql優化之精髓
- 為了不復制貼上,我被逼著學會了JAVA爬蟲
- 技術部突然宣佈:JAVA開發人員全部要會介面自動化測試框架
Redis 5種資料結構及對應使用場景,全會面試要加分的
寫在前邊
在家遠端辦公第三週,快被手機上的訊息搞的有些神經質了,生怕錯過一條有用的資訊,沒辦法形勢如此,公司搖搖欲墜大家也都如履薄冰,畢竟這時候失業有點慘(窮怕了)。
但就幹活來說還是比較清閒的,和在公司上班相比,清閒下來很多碎片時間,可以隨意的做點自己喜歡的事情。而且我發現,人一但閒下來真的是好可怕,潛在的才能會全面爆發,我女朋友這個抖音深度患者,一年不做一回飯的主,一週內接連給我做了兩頓黑暗料理,烤饅頭版“蛋糕”、漿糊版“涼皮”,然後我就與廁所結下來不解之緣。。。
不過,作為一個程式設計師,我對黑暗料理是不太感興趣滴,閒下來還是喜歡學習鑽研一些新奇的技術,canal
就成了很好的研究物件,一不小心就監控了公司MySQL的一舉一動的
一、canal是個啥?
canal
是阿里開發的一款基於資料庫增量日誌解析,提供增量資料訂閱與消費的框架,整個框架純JAVA
開發,目前僅支援Mysql
和MariaDB
(和mysql類似)。
那什麼是資料庫增量日誌?
MySQL的日誌種類是比較多的,主要包含:錯誤日誌、查詢日誌、慢查詢日誌、事務日誌、二進位制日誌。而MySQL
資料庫所發生的資料變更(DML
(data manipulation language)資料操縱語言,也就是我們熟悉的增刪改),都會以二進位制日誌(binary log
二、canal原理
在介紹canal
原理之前,我們先來回顧一下MySQL
主從同步的原理,這或許會讓你更好的理解canal
的工作機制。
1、MySQL主從同步原理:
MySQL主從同步也叫讀寫分離,可以提升資料庫的負載和容錯能力,實現資料庫的高可用
先來分析一張MySQL主從同步原理圖:
以上圖片源自網路,如有侵權聯絡刪除
master節點操作過程:
當master
節點資料發生更改後(delete、update、insert,還是建立函式、儲存過程等操作),向binary log
中寫入記錄日誌,這些記錄又叫做二進位制日誌事件
(binary log events)。
show binlog events
這些事件會按照順序寫入bin log中。當slave節點啟動連線到master節點的時候,master節點會為slave節點開啟binlog dump執行緒(負責傳輸binlog資料)。
一旦master節點的bin log發生變化時,bin log` `dump執行緒會通知slave節點有可以傳輸的binlog,並將相應的bin log內容傳送給slave節點。
slave節點操作過程:
slave節點上會建立兩個執行緒:一個I/O執行緒,一個SQL執行緒。I/O執行緒連線到master節點,master節點上的binlog dump
執行緒會將binlog的內容傳送給該I\O執行緒。
該I/O執行緒接收到binlog內容後,再將內容寫入到本地的relay log。而sql執行緒讀取到I/O執行緒寫入的ralay log,將relay log中的內容寫入slave資料庫。
2、canal原理
懂了上邊MySQL的主從同步原理,canal的工作機制就很好理解了。其實canal是模擬了MySQL資料庫中,slave節點與master節點的互動協議,偽裝自己為MySQL slave節點,向MySQL master節點發送dump協議
,MySQL master節點收到dump請求,開始推送binary log給slave節點(也就是canal
)。
以上圖片源自網路,如有侵權聯絡刪除
光說不練假把式,開幹!
三、canal實現“監控”MySQL
在寫程式碼前我們先對MySQL進行一下改造,安裝MySQL就不再細說了,基本操作。
1、檢視一下MySQL是否開啟了binary log功能
show binary logs
如果沒有開啟是圖中的狀態,一般使用者是沒有這個命令許可權的,不過我有,嘖嘖嘖!
如果沒有需要手動開啟,並且在my.cnf
檔案中配置binlog-format
為Row
模式
log-bin=mysq-bin
binlog-format=Row
log-bin
是binlog
檔案存放位置
binlog-format
設定MySQL複製log-bin的方式
MySQL的三種複製方式:
基於SQL語句的複製(statement-based replication, SBR)
- 優點:將修改資料的sql儲存在binlog,不需要記錄每一條sql和資料變化,binlog體量會很小,IO開銷少,效能好
- 缺點:會導致master-slave中的資料不一致
基於行的複製(row-based replication, RBR)
- 優點:不記錄每條sql語句的上下文資訊,僅需記錄哪條資料被修改了,修改成什麼樣了
- 缺點:binlog體積很大,尤其是在alter table屬性時,會產生大量binlog資料
混合模式複製(mixed-based replication, MBR)
- 對應的,binlog的格式也有三種:STATEMENT,ROW,MIXED。
2、為canal 建立一個有許可權操作MySQL的使用者
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;
3、安裝canal
下載地址:https://github.com/alibaba/canal/releases
下載後選擇版本例如:canal.deployer-xxx.tar.gz
4、配置canal
修改instance.properties檔案,需要新增監聽資料庫和表的規則,canal可以全量監聽資料庫,也可以針對某個表進行監聽,比較靈活。
vim conf/example/instance.properties
#################################################
## mysql serverId
canal.instance.mysql.slaveId = 2020
# position info 修改自己的資料庫(canal要監聽的資料庫 地址 )
canal.instance.master.address = 127.0.0.1:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
# username/password 修改成自己 資料庫資訊的賬號 (單獨開一個 準備階段建立的賬號)
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
# table regex 表的監聽規則
# canal.instance.filter.regex = blogs\.blog_info
canal.instance.filter.regex = .\*\\\\..\*
# table black regex
canal.instance.filter.black.regex =
啟動canal
sh bin/startup.sh
看一下server日誌,確認一下canal是否正常啟動
vi logs/canal/canal.log
顯示canal server is running now即為成功
2020-01-08 15:25:33.361 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2020-01-08 15:25:33.468 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.12.245:11111]
2020-01-08 15:25:34.061 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......
5、編寫Java客戶端程式碼,實現canal監聽
引入依賴包
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.0</version>
</dependency>
這裡只是簡單實現
public class MainApp {
public static void main(String... args) throws Exception {
/**
* 建立與
*/
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
11111), "example", "", "");
int batchSize = 1000;
int emptyCount = 0;
try {
connector.connect();
/**
* 監控資料庫中所有表
*/
connector.subscribe(".*\\..*");
/**
* 指定要監控的表,庫名.表名
*/
//connector.subscribe("xin-master.jk_order");
connector.rollback();
//120次心跳過後未檢測到,跳出
int totalEmptyCount = 120;
while (emptyCount < totalEmptyCount) {
Message message = connector.getWithoutAck(batchSize); // 獲取指定數量的資料
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
emptyCount++;
System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
printEntry(message.getEntries());
}
/**
* 提交確認
*/
connector.ack(batchId);
/**
* 處理失敗, 回滾資料
*/
connector.rollback(batchId);
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
/**
* 手動開啟事務回滾
*/
//TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
}
}
private static void printEntry(List<CanalEntry.Entry> entrys) {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry
.EntryType
.TRANSACTIONEND) {
continue;
}
CanalEntry.RowChange rowChage = null;
try {
rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
CanalEntry.EventType eventType = rowChage.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
if (eventType == CanalEntry.EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == CanalEntry.EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<CanalEntry.Column> columns) {
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
程式碼到這就編寫完成了,我們啟動服務看下是什麼效果,由於並沒有操作資料庫,所以監聽的結果都是空的。
接下來我們在資料庫執行一條update
語句試試
update jk_orderset order_no = '1111' where id = 40
控制檯檢測到了資料庫的修改,並生成binlog 日誌檔案mysql-bin.000009:3830
那麼生成的binlog 檔案該怎麼用,如何解析成SQl語句呢?
<!-- mysql binlog解析 -->
<dependency>
<groupId>com.github.shyiko</groupId>
<artifactId>mysql-binlog-connector-java</artifactId>
<version>0.13.0</version>
</dependency>
將剛才的binlog檔案下載本地測試一下
public static void main(String[] args) throws IOException {
String filePath = "C:\\ProgramData\\MySQL\\MySQL Server 5.7\\Data\\mysql-bin.000009:3830";
File binlogFile = new File(filePath);
EventDeserializer eventDeserializer = new EventDeserializer();
eventDeserializer.setChecksumType(ChecksumType.CRC32);
BinaryLogFileReader reader = new BinaryLogFileReader(binlogFile, eventDeserializer);
try {
for (Event event; (event = reader.readEvent()) != null; ) {
System.out.println(event.toString());
}
} finally {
reader.close();
}
}
檢視一下執行結果,發現數據庫最近的一次操作是加了一個idx_index索引
Event{header=EventHeaderV4{timestamp=1551325542000, eventType=ANONYMOUS_GTID, serverId=1, headerLength=19, dataLength=46, nextPosition=8455, flags=0}, data=null}
Event{header=EventHeaderV4{timestamp=1551325542000, eventType=QUERY, serverId=1, headerLength=19, dataLength=190, nextPosition=8664, flags=0}, data=QueryEventData{threadId=25, executionTime=0, errorCode=0, database='xin-master', sql='ALTER TABLE `jk_order`
DROP INDEX `idx_index` ,
ADD INDEX `idx_index` (`user_id`, `service_id`, `real_price`) USING BTREE'}}
Event{header=EventHeaderV4{timestamp=1551438586000, eventType=STOP, serverId=1, headerLength=19, dataLength=4, nextPosition=8687, flags=0}, data=null}
至此我們就已經實現了監控MySQL,
四、canal應用場景
canal
應用場景大致有以下:
- 解決MySQL主從同步延遲的問題
- 實現資料庫實時備份
- 多級索引 (賣家和買家各自分庫索引)
- 實現業務cache重新整理
- 價格變化等重要業務訊息
重點分析一下canal是如何解決MySQL主從同步延遲的問題
生產環境下MySQL
的主從同步模式(maser-slave
)很常見,但對於跨機房部署的叢集,會出現同步延時的情況。舉個栗子:
一條訂單狀態是未付款,master
節點修改成已付款,可由於某些原因出現延遲資料未能及時同步到slave
,這時使用者立即檢視訂單狀態(查詢走slave
)顯示還是未付款,哪個使用者看到這種情況不得慌啊。
為什麼會出現主從同步延遲呢?
當主庫master
的TPS
併發較高時,master
節點併發產生的修改操作,而slave
節點的sql執行緒
是單執行緒處理同步資料,延時自然而言就產生了。
不過造成主從同步的原因不止這些,由於主從伺服器存在跨機器並且跨機房,除了網路頻寬原因之外,網路的穩定性以及機器之間的同步,都是主從同步應該考慮的主要原因。
總結
本文只是簡單實現canal監聽資料庫的功能,旨在給大家提供一種解決問題的思路,還是反覆絮叨的那句話,解決問題的技術方法很對,具體如何應用還需結合具體業務。
今天就說這麼多,如果本文對您有一點幫助,希望能得到您一個點贊