注册    登录
关注
Py学习  »  DATABASE

用Python开发MySQL增强半同步BinlogServer（T2通信篇）

老叶茶馆 • 4 年前 • 373 次点击  
导读：
作者：曾永伟，知数堂10期学员，多年JAVA物流行业开发管理经验和PHP/Python跨境电商开发管理经验，对数据库系统情有独钟，善于运用SQL编程简化业务逻辑，去年开始正式从业MySQL DBA, 专注于DB系统自动化运维、MySQL云上实践。
本文为python-mysql-binlogserver系列的第二篇（T2通信篇），之后会陆续发布此系列的其他文章，请大家点击在看或者收藏，自行练习。

一想到你在关注我就忍不住有点紧张


一、概述

从前一篇文章我们已经了解了：

  • 二进制在计算机中的应用

  • 字符集与二进制的关系及其在Python中处理

  • 用struct来处理二进制的转换问题

  • Socket通信的基本编程方法

在本节中，我们将结合这些内容进一步来了解如何使Python和MySQL进行交互。


二、MySQL通信协议基础

在Python中，使用socket.secv接收数据或send发送数据的都是二进制流对象Bytes，我们需要结合MySQL通信协议来逐字节解析其具体的含义。

MySQL基础通信单位Packet，它由header + payload组成，header由3个字节的payload长度（最大16M字节数）和1个字节的流水号组成，在读取一个Packet时，通常先读4个字节，解析出payload的长度和payload的序号，再根据payload的长度把余下的正文读取出来。

  1. s = socket.


    
socket(socket.AF_INET, socket.SOCK_STREAM)
  2. s.connect(("127.0.0.1", 3306))
  3. '''
  4. # 先读Header
  5. header = s.recv(4)
  6. length = struct.unpack('<I', header[:3] + b'\x00')[0]
  7. sequenceId = struct.unpack('<B', header[:-1])[0]

  9. # 再读余下的正文，完成一个Packet的完整读取
  10. 


    
body = s.recv(length)

解析Greeting包

当Client连接上MySQL Server后，MySQL会主动发送一个greeting包，把自己的状态和随机密文发送给Client, 等待Client响应帐户和密码等信息，验证失败发送ERR包并主动断开连接，验证成功后发送OK包，保持连接，等待Client发送其它"指令"。

接下来我先看一下Greeting包正文的官方说明：

  1. https://dev.mysql.com/doc/internals/en/connection-phase-packets.html
  2. 1              [0a] protocol version
  3. string[NUL]    server version
  4. 4              connection id
  5. string[8]      auth-plugin-data-part-1
  6. 1              [


    
00] filler
  7. 2              capability flags (lower 2 bytes)
  8. 1              character set
  9. 2              status flags
  10. 2              capability flags (upper 2 bytes)
  11. 1              length of auth-plugin-data
  12. string[10]     reserved (all [00])
  13. 


    
string[$len]   auth-plugin-data-part-2 ($len=MAX(13, length of auth-plugin-data - 8))
  14. string[NUL]    auth-plugin name

为了更加直接的观察和理解，我这里去掉了对低版本的兼容格式，让其显得更加整洁。

我们来尝试用Python解析第一个Greeting包：

  1. import socket
  2. import struct
  3. from


    
 pprint import pprint

  5. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  6. s.connect(("192.168.56.101", 3306))

  8. greeting = {}

  10. # Header 3+1
  11. greeting["length"] = struct.unpack(',


    
 s.recv(3) + b'\x00')[0]
  12. greeting["sequenceId"] = struct.unpack(', s.recv(1))[0]

  14. # 正文开始
  15. greeting["protocolVersion"] = s.recv(1)

  17. # serverVersion是string[NUL]类型，所以一直循环读取到\x00节束
  18. 


    
greeting["serverVersion"] = ""
  19. while True:
  20.     _byte = s.recv(1)
  21.     if _byte == b'\x00':
  22.         break
  23.     greeting["serverVersion"] += chr(int(_byte.hex


    
(), 16))

  25. # 余下部分请自行参照上方文档进行一一对照
  26. greeting["connectionId"] = struct.unpack(', s.recv(4))[0]
  27. greeting["challenge1"] = s.recv(8).decode("utf8")
  28. _filler = s.recv(


    
1)
  29. greeting["capabilityFlags"] = s.recv(2)
  30. greeting["characterSet"] = s.recv(1)
  31. greeting["statusFlags"] = s.recv(2)
  32. greeting["capabilityFlag"] = s.recv(


    
2)
  33. greeting["authPluginDataLength"] = struct.unpack(', s.recv(1))[0]
  34. _reserved = s.recv(10)
  35. greeting["challenge2"] = s.recv(max(13, greeting["authPluginDataLength"] - 


    
8)).decode("utf8")
  36. greeting["authPluginName"] = ""
  37. while True:
  38.     _byte = s.recv(1)
  39.     if _byte == b'\x00':
  40.         break
  41.     greeting["authPluginName"] += chr


    
(int(_byte.hex(), 16))
  42. pprint(greeting)

输出：

  1. {'authPluginDataLength': 21,
  2.  'authPluginName': 'mysql_native_password',
  3.  'capabilityFlag': b'\xff\x81',
  4.  'capabilityFlags': b'\xff\xf7',
  5. 


    
 'challenge1': 'X:u8\x11N4\x1b',
  6.  'challenge2': 'dF\x04f~\x1f!%\x14\x1acV\x00',
  7.  'characterSet': b'\xe0',
  8.  'connectionId': 73,
  9.  'length': 78,
  10.  'protocolVersion': b'\n',
  11.  'sequenceId': 0


    
,
  12.  'serverVersion': '5.7.20-log',
  13.  'statusFlags': b'\x02\x00'}

可见用Python解析出Greeting包的内容并没有想象中那么难，只要结合官方文档，用Python struct很容易就把二进制流还原成可读的文本信息。

为了使代码可以复用,我们将上面的代码进行一个简单的函数化封装，以方便给后面的例子调用：

  1. # learn_packet1_greeting.py

  3. import socket
  4. import struct
  5. from pprint import pprint


  8. def get_greeting(s):
  9. 


    
    greeting = {}
  10.     greeting["length"] = struct.unpack(', s.recv(3) + b'\x00')[0]
  11.     greeting["sequenceId"] = struct.unpack(', s.recv(1))[0]
  12.     greeting


    
["protocolVersion"] = s.recv(1)
  13.     greeting["serverVersion"] = ""
  14.     while True:
  15.         _byte = s.recv(1)
  16.         if _byte == b'\x00':
  17.             break
  18.         greeting


    
["serverVersion"] += chr(int(_byte.hex(), 16))
  19.     greeting["connectionId"] = struct.unpack(', s.recv(4))[0]
  20.     greeting["challenge1"] = s.recv(8).decode("utf8")
  21. 


    
    _filler = s.recv(1)
  22.     greeting["capabilityFlags"] = s.recv(2)
  23.     greeting["characterSet"] = s.recv(1)
  24.     greeting["statusFlags"] = s.recv(2)
  25.     greeting[


    
"capabilityFlag"] = s.recv(2)
  26.     greeting["authPluginDataLength"] = struct.unpack(', s.recv(1))[0]
  27.     _reserved = s.recv(10)
  28.     greeting["challenge2"] = s.recv(max(13


    
, greeting["authPluginDataLength"] - 8)).decode("utf8")
  29.     greeting["authPluginName"] = ""
  30.     while True:
  31.         _byte = s.recv(1)
  32.         if _byte == b'\x00':
  33.             break
  34. 


    
        greeting["authPluginName"] += chr(int(_byte.hex(), 16))
  35.     return greeting

  37. if __name__ == "__main__":
  38.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  39.     s.connect(("192.168.1.100",


    
 3306))
  40.     greeting = get_greeting(s)
  41.     pprint(greeting)

封装Response包，完成验证

当Client得到Greeting包后，就可以结合两个随机码，组成认证回应包，完成MySQL的认证：

  1. # learn_packet2_auth.py

  3. import socket
  4. import struct
  5. import hashlib
  6. from functools import partial
  7. from py_mysql_binlogserver


    
._tutorial.learn_packet1_greeting import get_greeting


  10. def dump_packet(packet, title=None):
  11.     pass        # 省略代码，节约篇幅


  14. def scramble_native_password(password, message):
  15.     '''
  16.     mysql_native_password
  17.     https://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41
  18.     '''
  19.     SCRAMBLE_LENGTH = 


    
20
  20.     sha1_new = partial(hashlib.new, 'sha1')

  22.     """Scramble used for mysql_native_password"""
  23.     if not password:
  24.         return b''

  26.     password = password.encode("utf-8")
  27.     message = message.encode("utf-8")

  29. 


    
    stage1 = sha1_new(password).digest()
  30.     stage2 = sha1_new(stage1).digest()
  31.     s = sha1_new()
  32.     s.update(message[:SCRAMBLE_LENGTH])
  33.     s.update(stage2)
  34.     result = s.digest()
  35. 


    
    return _my_crypt(result, stage1)


  38. def _my_crypt(message1, message2):
  39.     result = bytearray(message1)
  40.     for i in range(len(result)):
  41.         result[i] ^= message2[i]

  43.     return


    
 bytes(result)

  45. def get_response(s, username, password, challenge1, challenge2):
  46.     '''
  47.     https://dev.mysql.com/doc/internals/en/connection-phase-packets.html
  48.     简化版
  49.     4              capability flags, CLIENT_PROTOCOL_41 always set
  50.     4              max-packet size
  51.     1              character set
  52.     string[23]     reserved (all [0])
  53.     string[NUL]    username
  54. 


    
    lenenc-int     length of auth-response
  55.     string[n]      auth-response
  56.     string[NUL]    auth plugin name
  57.     '''
  58.     scramble_password = scramble_native_password(password, challenge1 + challenge2)

  60.     response = b''
  61.     response += struct.pack(', 32482821)
  62.     response += struct.pack('


    
, 16777216)
  63.     response += struct.pack(', 33)
  64.     response += b''.join([b'\x00' for i in range(23)])
  65.     response += username.encode() + b'\x00'
  66.     response += struct.


    
pack(', len(scramble_password))
  67.     response += scramble_password
  68.     response += "mysql_native_password".encode() + b'\x00'
  69.     response = struct.pack(', len(response))[:-1] + struct.pack(', 1) + response


    

  70.     return response


  73. if __name__ == "__main__":

  75.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  76.     s.connect(("192.168.1.100", 3306))

  78.     greeting = get_greeting(s)

  80.     username =


    
 'repl'
  81.     password = 'repl1234'
  82.     response = get_response(s, username, password, greeting["challenge1"], greeting["challenge2"])
  83.     s.send(response)
  84.     dump_packet(response, "Response packet:")

  86.     result = s.recv(1024)
  87. 


    
    dump_packet(result, "Result packet:")

向服务器发送帐号及加密的密文，通过验证后，会得到OK包：

  1. Response packet:
  2. 00000000  50 00 00 01 05 A6 EF 01  00 00 00 01 21 00 00 00   P....¦ï. ....!...
  3. 00000010


    
  00 00 00 00 00 00 00 00  00 00 00 00 00 00 00 00   ........ ........
  4. 00000020  00 00 00 00 72 65 70 6C  00 14 F6 4A


    
 C1 E7 4F 2A   ....repl ..öJÁçO*
  5. 00000030  BB A3 CB 29 3C 5B 50 F9  3C AF E3 6C 1C A9 6D 79   »£Ë)Pù l.©my
  6. 00000040  73 71 6C 5F 6E 61 74


    
 69  76 65 5F 70 61 73 73 77   sql_nati ve_passw
  7. 00000050  6F 72 64 00                                       ord.

  9. Result packet:
  10. 00000000  07 00 00 02 00 00 00 02  00


    
 00 00                  ........ ...

验证失败会得到如下结果：

  1. Response packet:
  2. 00000000  50 00 00 01 05 A6 EF 01  00 00 00 01 21 00 00 00   P....¦ï. ....!...
  3. 00000010  00 


    
00 00 00 00 00 00 00  00 00 00 00 00 00 00 00   ........ ........
  4. 00000020  00 00 00 00 72 65 70 6C  00 14 AB 41 90 


    
72 07 98   ....repl ..«A�r.�
  5. 00000030  0A 36 21 F8 FC CC 83 7B  8E 4E A3 75 A2 DB 6D 79   .6!øüÌ�{ �N£u¢Ûmy
  6. 00000040  73 71 6C 5F 


    
6E 61 74 69  76 65 5F 70 61 73 73 77   sql_nati ve_passw
  7. 00000050  6F 72 64 00                                       ord.

  9. Result packet:
  10. 00000000  4A 00 00 02 FF 15 04 


    
23  32 38 30 30 30 41 63 63   J......# 28000Acc
  11. 00000010  65 73 73 20 64 65 6E 69  65 64 20 66 6F 72 20 75   ess deni ed for u
  12. 


    
00000020  73 65 72 20 27 72 65 70  6C 27 40 27 31 39 32 2E   ser 'rep l'@'192.
  13. 00000030  31 36 38 2E 31 2E 31 27  20 28 75 73 69 6E 67 20   168.1.1'  (using 
  14. 00000040  70 61 73 73 77


    
 6F 72 64  3A 20 59 45 53 29         password : YES)

执行查询

完成认证后，服务器就会保存连接，直到超时或Client主动退出才会中断连接。接下来我们就可以在完成认证后的socket上发送命令了，就像使用mysql client一样，只不过通过socket发送的数据还要是bytes对象，直接上菜：

  1. # learn_packet3_query.py

  3. import socket
  4. import struct

  6. from py_mysql_binlogserver._tutorial.learn_packet1_greeting import get_greeting
  7. from py_mysql_binlogserver


    
._tutorial.learn_packet2_auth import dump_packet, get_response


  10. def read_packet(skt):
  11.     while True:
  12.         _header = skt.recv(5)
  13.         _length = struct.unpack(", (_header[0:3] + b"\x00"))[


    
0]
  14.         _sequenceId = struct.unpack(", _header[3:4])[0]
  15.         _packetType = struct.unpack(", _header[4:])[0]

  17.         # 每个查询会产生多个数据包，读到 EOF 包后结束
  18.         if _packetType == 0xfe:     # EOF
  19. 


    
            break

  21.         _payload = skt.recv(_length - 1)
  22.         dump_packet(_header + _payload, f"read packet [{_sequenceId}]")


  25. def get_query(sql):
  26.     """
  27.     https://dev.mysql.com/doc/internals/en/com-query.html
  28.     1              [03] COM_QUERY
  29.     string[EOF]    the query the server shall execute
  30.     """
  31. 


    
    query = b''
  32.     query += struct.pack(", 3)
  33.     query += sql.encode()

  35.     query = struct.pack(', len(query))[:-1] + struct.pack(', 0)


    
 + query
  36.     return query


  39. if __name__ == "__main__":

  41.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  42.     s.connect(("192.168.1.100", 3306))

  44.     greeting = get_greeting(s)
  45. 


    
    username = 'repl'
  46.     password = 'repl1234'
  47.     response = get_response(s, username, password, greeting["challenge1"], greeting["challenge2"])
  48.     s.send(response)
  49.     result = s.recv(1024)

  51.     sql = "select @@version_comment"
  52. 


    
    query = get_query(sql)
  53.     dump_packet(query, f"query packet:{sql}")
  54.     s.send(query
    )


  56.     read_packet(s)


三、Binlog文件格式
先简单回顾一下binlog文件中都有些什么内容：
  1. mysql> show binlog events in 'mysql-bin.000009';
  2. +------------------+-----+----------------+-----------+-------------+--------------------------------------------------------------------+
  3. 


    
| Log_name         | Pos | Event_type     | Server_id | End_log_pos | Info                                                               |
  4. +------------------+-----+----------------+-----------+-------------+--------------------------------------------------------------------+
  5. | mysql-bin.000009 |   4 | Format_desc    |   3306100 |         123 | 


    
Server ver: 5.7.20-log, Binlog ver: 4                              |
  6. | mysql-bin.000009 | 123 | Previous_gtids |   3306100 |         190 | f0ea18e0-3cff-11e9-9488-0800275ae9e7:1-19                          |
  7. 


    
| mysql-bin.000009 | 190 | Gtid           |   3306100 |         251 | SET @@SESSION.GTID_NEXT= 'f0ea18e0-3cff-11e9-9488-0800275ae9e7:20' |
  8. | mysql-bin.000009 | 251 | Query          |   3306100 |         318


    
 | BEGIN                                                              |
  9. | mysql-bin.000009 | 318 | Table_map      |   3306100 |         361 | table_id: 223 (db3.t3)                                             |
  10. | mysql-bin.000009 | 361 |


    
 Write_rows     |   3306100 |         402 | table_id: 223 flags: STMT_END_F                                    |
  11. | mysql-bin.000009 | 402 | Xid            |   3306100 |         429 | COMMIT /* xid=286 */                                               |
  12. | mysql-bin.000009


    
 | 429 | Gtid           |   3306100 |         490 | SET @@SESSION.GTID_NEXT= 'f0ea18e0-3cff-11e9-9488-0800275ae9e7:21' |
  13. | mysql-bin.000009 | 490 | Query          |   3306100 |         557 | BEGIN                                                              |
  14. 


    
| mysql-bin.000009 | 557 | Table_map      |   3306100 |         600 | table_id: 223 (db3.t3)                                             |
  15. | mysql-bin.000009 | 600 | Write_rows     |   3306100 |         


    
641 | table_id: 223 flags: STMT_END_F                                    |
  16. | mysql-bin.000009 | 641 | Xid            |   3306100 |         668 | COMMIT /* xid=287 */                                               |
  17. | mysql-bin.000009 | 668 | Rotate         


    
|   3306100 |         711 | mysql-bin.000010;pos=4                                             |
  18. +------------------+-----+----------------+-----------+-------------+--------------------------------------------------------------------+
  19. 13 rows in set (0.00 sec)

可以看出，binlog文件内容就是有很多的Event组成，一个完整的binlog应该是由Format_desc event开始，Rotate event结束，它们充当Binlog文件的元数据，中间Event才是真正和数据相关Event,每一个Event的格式都不尽相同，需要单独作解析。不过我们的BinlogServer并不关心具体的Event内容，只需要把Event作为一个接收，存储和发送的基本单元即可，简单说就是，把Master发送的Event按顺序存储起来，当有Slave change过来以后，再从指定位置把Event一个一个的发送给Slave，仅此而已。

每一个Event都有自己的Headyier, 描述了它的创建时间、类型、Server Id、长度、下一个Event位置和flags信息，结合Event的长度和下个Event位置信息，我们可以很容易地实现顺序扫描一个Binlog文件中的所有Event：

  1. # learn_bin2_binlog.py

  3. import struct
  4. from


    
 py_mysql_binlogserver.constants.EVENT_TYPE import event_type_name

  6. event_map = event_type_name()

  8. with open("mysql-bin.000009", mode="rb") as fr:
  9.     _file_header = fr.read(4)
  10.     if _file_header != bytes.fromhex("fe62696e"):
  11.         print(


    
"It is not a binlog file.")
  12.         exit()

  14.     '''
  15.     https://dev.mysql.com/doc/internals/en/binlog-event-header.html
  16.     4              timestamp
  17.     1              event type
  18.     4              server-id
  19.     4              event-size
  20.     4              log pos
  21.     2              flags
  22.     '''
  23.     while True:
  24. 


    
        event_header = fr.read(19)
  25.         if len(event_header) == 0:
  26.             break
  27.         timestamp, event_type, server_id, event_size, log_pos, flags = struct.unpack(', event_header)
  28.         print("Binlog Event[%s]: [%s] %s %s" %


    
 (timestamp,
  29.                                                 event_type,
  30.                                                 event_map.get(event_type), log_pos))
  31.         event_body = fr.read(event_size - 19)

输出：

  1. Binlog Event[1570889546]: [15] FORMAT_DESCRIPTION_EVENT 123
  2. Binlog Event


    
[1570889546]: [35] PREVIOUS_GTIDS_LOG_EVENT 190
  3. Binlog Event[1570889807]: [33] GTID_LOG_EVENT 251
  4. Binlog Event[1570889807]: [2] QUERY_EVENT 318
  5. Binlog Event[1570889807]: [19] TABLE_MAP_EVENT 361
  6. Binlog Event


    
[1570889807]: [30] WRITE_ROWS_EVENT 402
  7. Binlog Event[1570889807]: [16] XID_EVENT 429
  8. Binlog Event[1570889813]: [33] GTID_LOG_EVENT 490
  9. Binlog Event[1570889813]: [2] QUERY_EVENT 557
  10. Binlog Event[


    
1570889813]: [19] TABLE_MAP_EVENT 600
  11. Binlog Event[1570889813]: [30] WRITE_ROWS_EVENT 641
  12. Binlog Event[1570889813]: [16] XID_EVENT 668
  13. Binlog Event[1570889820]: [4] ROTATE_EVENT 711

是不是比想象中简单，如果你把基础篇的内容全部理解了，我相信上面这段代码不会难到你，相反如果你还不能理解上面这段代码，请移步回去多看几遍，多练几遍再回来。更多的Binlog相关知识，请参考官方文档。


四、小结

这一节我们把Socket通信的难关攻破了，已经完成了和MySQL服务器进行握手，登陆和执行查询，近距离的接触了MySQL的通信协议，学会了如何运用Python struct进行简单的解包和封包，也简单分析了binlog的组成及使用Python来解析binlog文件，最重要的是学会了结合官方文档来解决我们的实际问题。

下一节我们将在本节的基础上进入实战篇，利用Socket向Master发起Slave注册，发送BinlogDump指令，以获取Master上的Binlog Event并保存到本地文件中。


相关文档

  • https://dev.mysql.com/doc/internals/en/mysql-packet.html

  • https://dev.mysql.com/doc/internals/en/binlog-file-header.html

  • https://dev.mysql.com/doc/internals/en/binlog-event-header.html

  • https://github.com/alvinzane/py-mysql-binlogserver/tree/master/pymysqlbinlogserver/_tutorial

（本周先为大家献上这两篇文章，大家可以多练习，下周会继续发布此系列的文章，请持续关注哦~~）



扫码加入MySQL技术Q群

（群号：650149401）

   

点“在看”给我一朵小黄花



Python社区是高质量的Python/Django开发社区
本文地址：http://www.python88.com/topic/50340
Reply
 
373 次点击  
登录后回复
关于   移动版   ·   三行代码   ·   今天看啥   ·   Code   ·   link之家   ·   卧龙搜索   ·   藏经阁   ·   小百科
Py学习 - 专注于Python技术发展的社区(原Django社区)
沪ICP备11025650号