MySQL系列：基于binlog的增量订阅与消费（一）

MySQL授权

1 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘replicator‘@‘192.168.3.%‘ IDENTIFIED BY ‘123456‘;

Binlog要满足如下条件

 1 MySQL>root@(none) 09:53:38>show variables like ‘log_bin‘;
 2 +---------------+-------+
 3 | Variable_name | Value     |
 4 +---------------+-------+
 5 | log_bin       | ON       |
 6 +---------------+-------+
 7 1 row in set (0.01 sec)
 8 
 9 MySQL>show variables like ‘binlog_format‘;
10 +---------------+-------+
11 | Variable_name | Value     |
12 +---------------+-------+
13 | binlog_format | ROW      |
14 +---------------+-------+
15 1 row in set (0.00 sec)
16 
17 MySQL>show variables like ‘binlog_row_image‘;
18 +------------------+-------+
19 | Variable_name    | Value     |
20 +------------------+-------+
21 | binlog_row_image | FULL      |
22 +------------------+-------+
23 1 row in set (0.00 sec)

示例代码：

 1 [root@mha-maxscale-1 script]# cat mysql-replication.py
 2 #!/usr/bin/env python
 3 # -*- coding: utf-8 -*-
 4 
 5 from pymysqlreplication import BinLogStreamReader
 6 from pymysqlreplication.row_event import (
 7     DeleteRowsEvent,
 8     UpdateRowsEvent,
 9     WriteRowsEvent,
10 )
11 import sys
12 import json
13 
14 def main():
15     mysql_settings = {‘host‘: ‘192.168.3.130‘,
16                       ‘port‘: 3306, ‘user‘: ‘replicator‘, ‘passwd‘: ‘123456‘}
17     stream = BinLogStreamReader(
18         connection_settings=mysql_settings,
19         server_id=101,
20         blocking=True,
21         only_schemas=[‘zow‘],
22         only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
23         resume_stream=True,
24         log_file=‘mysql-bin.000013‘, log_pos=6197)
25 
26     for binlogevent in stream:
27         for row in binlogevent.rows:
28             event = {"schema": binlogevent.schema, "table": binlogevent.table, "log_pos": binlogevent.packet.log_pos}
29             if isinstance(binlogevent, DeleteRowsEvent):
30                 event["action"] = "delete"
31                 event["values"] = dict(row["values"].items())
33                 event = dict(event.items())
34             elif isinstance(binlogevent, UpdateRowsEvent):
35                 event["action"] = "update"
36                 event["before_values"] = dict(row["before_values"].items())
37                 event["after_values"] = dict(row["after_values"].items())
38                 event = dict(event.items())
39             elif isinstance(binlogevent, WriteRowsEvent):
40                 event["action"] = "insert"
41                 event["values"] = dict(row["values"].items())
42                 event = dict(event.items())
43             print json.dumps(event)
44             sys.stdout.flush()
45 
46     stream.close()
47 
48 
49 if __name__ == "__main__":
50     main()

运行结果：

1 [root@mha-maxscale-1 script]# python mysql-replication.py 
2 {"action": "insert", "table": "t2", "log_pos": 4622, "values": {"tname": "hh", "id": 7}, "schema": "zow"}
3 {"log_pos": 4904, "after_values": {"tname": "ii", "id": 7}, "action": "update", "table": "t2", "before_values": {"tname": "hh", "id": 7}, "schema": "zow"}
4 {"log_pos": 4904, "after_values": {"tname": "ii", "id": 7}, "action": "update", "table": "t2", "before_values": {"tname": "hh", "id": 7}, "schema": "zow"}
5 {"action": "delete", "table": "t2", "log_pos": 5169, "values": {"tname": "ii", "id": 7}, "schema": "zow"}
6 {"action": "delete", "table": "t2", "log_pos": 5169, "values": {"tname": "ii", "id": 7}, "schema": "zow"}

更多例子见：https://github.com/noplay/python-mysql-replication/tree/master/examples

MySQL系列：基于binlog的增量订阅与消费（一）

标签：packet   write   mys   基于   before   rip   mysql数据库   ict   3.5