用Python开发MySQL增强半同步BinlogServer（T1基础篇）

作者&＃xff1a;曾永伟&＃xff0c;知数堂10期学员&＃xff0c;多年JAVA物流行业开发管理经验和PHP/Python跨境电商开发管理经验&＃xff0c;对数据库系统情有独钟&＃xff0c;善于运用SQL编程简化业务逻辑&＃xff0c;去年开始正式从业MySQL DBA, 专注于DB系统自动化运维、MySQL云上实践。

本文为python-mysql-binlogserver系列的第一篇&＃xff08;T1基础篇&＃xff09;&＃xff0c;之后会陆续发布此系列的其他文章&＃xff0c;请大家点击在看或者收藏&＃xff0c;自行练习。

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

概述

前不久知数堂吴老师在公开课上把MHA拉下神坛&＃xff0c;是因为MHA无法从根本上解决丢数据的可能性&＃xff0c;只是尝试性的去补偿未同步的数据。使用MySQL的半同步复制可以解决数据丢失的问题&＃xff0c;但原生 io_thread会破坏掉Master上的Binlog File的命名&＃xff0c;对后继的运维造成不便&＃xff0c;所以使用原生增强半同步&＃43;blackhole引擎做binlog备份的方案几乎没有人使用&＃xff0c;而更多的公司则使用mysqlbinlog命令来实现轻量的BinlogServer&＃xff0c;不足的是官方mysqlbinlog并不支持半同步复制&＃xff0c;仍然会丢数据。

据我所知&＃xff0c;Facebook,Google和国内的美团公司都有研发自己的BinlogServer&＃xff0c;但是目前我没有找到一个开源的支持半同步的BinlogServer项目&＃xff0c;于是就诞生了py-mysql-binlogserver这个项目。

1、主要特性如下&＃xff1a;

• 全Python标准模块开发&＃xff0c;无第三方库依赖&＃xff0c;减少学习成本
  

• 独立Dumper进程&＃xff0c;用于同步保存Binlog event
  

• 支持半同步协议&＃xff0c;数据零丢失

• 独立Server进程&＃xff0c;支持Failover时Change master to来补数据
  

• 支持GTID&＃xff0c;方便切换Master
  

• 暂不支持级联复制模式
  

• 仅在MySQL官方版5.7&＃43;下测试通过
  

• 仅支持Python3, 不兼容Python2

2、TODO特性

• 使用mysql client协议来管理binlogserver

• 实现状态信息查询&＃xff0c;满足监控需求

具体功能请到项目首页进行查看和下载体验。

3、项目地址为&＃xff1a;

https://github.com/alvinzane/py-mysql-binlogserver

4、目录结构&＃xff1a;

1. py-mysql-binlogserver

2. ├── README.doc

3. ├── doc

4. │ ├── T1基础篇-用Python开发MySQL增强半同步BinlogServer.md

5. │ ├── T2通信篇-用Python开发MySQL增强半同步BinlogServer.md

6. │ ├── T3实战篇-用Python开发MySQL增强半同步BinlogServer.md

7. │ ├── T4架构篇-用Python开发MySQL增强半同步BinlogServer.md

8. │ └── readme.md

9. └── py_mysql_binlogserver

10. ├── __init__.py

11. ├── _playground # 练习场&＃xff0c;随便玩

12. │ ├── __init__.py

13. │ ├── socket_client.py

14. │ ├── socket_client_semi-repl.py

15. │ └── test_slave.py

16. ├── _tutorial # 教程实例代码

17. │ ├── __init__.py

18. │ ├── learn_bin1_charset.py

19. │ ├── learn_bin2_binlog.py

20. │ ├── learn_packet1_greeting.py

21. │ ├── learn_packet2_auth.py

22. │ ├── learn_packet3_query.py

23. │ ├── learn_packet4_dump.py

24. │ ├── learn_packet4_dump2.py

25. │ ├── learn_packet5_dump_with_semi_ack.py

26. │ ├── learn_socket1_client.py

27. │ ├── learn_socket2_server.py

28. │ ├── learn_socket3_server_mulit_thread.py

29. │ └── learn_socket4_server_mulit_thread.py

30. ├── binlogs # Binlog文件保存目录

31. │ ├── mysql-bin.000014

32. │ ├── mysql-bin.gtid.index

33. │ └── mysql-bin.index

34. ├── cap

35. ├── constants

36. │ ├── EVENT_TYPE.py

37. │ └── FIELD_TYPE.py

38. ├── dump

39. │ └── readme.md

40. ├── packet

41. │ ├── __init__.py

42. │ ├── binlog_event.py

43. │ ├── challenge.py

44. │ ├── dump_gtid.py

45. │ ├── dump_pos.py

46. │ ├── event_header.py

47. │ ├── gtid_event.py

48. │ ├── query.py

49. │ ├── response.py

50. │ ├── semiack.py

51. │ └── slave.py

52. ├── protocol

53. │ ├── Flags.py

54. │ ├── __init__.py

55. │ ├── err.py

56. │ ├── gtid.py

57. │ ├── ok.py

58. │ ├── packet.py

59. │ └── proto.py

60. └── tests # 单元测试

61. │ ├── __init__.py

62. │ └── test_packet.py

63. ├── proxy.py # 简单代理&＃xff0c;用于观察和保存MySQL Packet

64. ├── server.py # 实现Master协议

65. ├── dumper.py # Binlog Dumper,相当于IO Thread

66. ├── example.conf # 配置文件

67. └─── example.py # 同时启动 Server&Dumper

假设你已经有了一定的Python编程基础&＃xff0c;并且完全理解MySQL半同步复制的原理&＃xff0c;接下来我们就一步一步走进实现BinlogServer的技术细节。

二进制基础复习

MySQL的Binlog文件是二进制的&＃xff0c;MySQL在网络上传送的数据也是二进制的&＃xff0c;所以我们先来复习一下二进制的一些基础知识。

1、数字的表示&＃xff1a;

数字通常可以用十进制&＃xff0c;二进制和十六进制来表示。在计算机中&＃xff0c;数据的运算、存储、传输最终都会用到二进制&＃xff0c;但由于二进制不便于人类阅读&＃xff08;太长了&＃xff09;&＃xff0c;所以我们通常用一位十六进制来表示四个bite的二进制&＃xff0c;即2位十六制表示一个Byte(字节)。

1. 二进制 十六进制 十进制

2. 0000 0001 01 1

3. 0000 0010 02 2

4. ...

5. 0000 1010 0A 10

6. 1111 1111 FF 255

在Python中&＃xff0c;用非零开头的数字&＃xff0c;表示十进制&＃xff0c;0x,0b开头分别表示十六进制和二进制&＃xff1a;

1. # 数字10的三种表示&＃xff1a;

2. >>> print(10,0xa,0b1010)

3. 10 10 10

十六进制和二进制与十进制转换&＃xff1a;

1. >>> print(hex(10),bin(10))

2. 0xa 0b1010

3. >>> print(int(&＃39;0xa&＃39;,16),int(&＃39;0b1010&＃39;,2))

4. 10 10

由于1个字节&＃xff08;byte)最大能表示的数字为255&＃xff0c;所以更大的数字需要用多个字节来表示&＃xff0c;如&＃xff1a;

1. # 2个字节&＃xff0c;16bit,最大为 65535

2. >>> 0b1111111111111111

3. 65535

5. # 4个字节&＃xff0c;32bit, 最大数&＃xff08;0x为十六进制&＃xff0c;1位十六进制等于4位二进制 0xf &＃61; 0b1111

6. >>> 0xffffffff

7. 4294967295

以上均为无符号的数字&＃xff0c;即全为正数&＃xff0c;对于有符号的正负数&＃xff0c;则最高位的1个bit用0和1分别表示正数和负数。对于1个byte的数字&＃xff0c;实际就只有7bit表示实际的数字&＃xff0c;范围为[-128&＃xff0c;127].

2、字符的表示

在计算机中所有的数据最终都要转化为数字&＃xff0c;而且是二进制的数字。字符也不例外&＃xff0c;也需要用到一个"映射表"来完成字符的表示。这个"映射表"叫作字符集&＃xff0c;ASCII是最早最基础的"单字节"字符集&＃xff0c;它可以表示键盘上所有的可打印字符&＃xff0c;如52个大小写字母及标点符号。

Python中&＃xff0c;使用ord()和chr()完成ASCII字符与数字之间的转换&＃xff1a;

1. >>> ord(&＃39;a&＃39;),ord(&＃39;b&＃39;),ord(&＃39;c&＃39;)

2. (97, 98, 99)

3. >>> chr(97),chr(98),chr(99)

4. (&＃39;a&＃39;, &＃39;b&＃39;, &＃39;c&＃39;)

"单字节"最大为数字是255&＃xff0c;能表示的字符有限&＃xff0c;所以后来就有了"多字节"字符集&＃xff0c;如GBK&＃xff0c;UTF8等等&＃xff0c;用来表示更多的字符。其中UTF8是变长的字符编码&＃xff0c;用1-6个字节表示一个字符&＃xff0c;可以表示全世界所有的文字与符号&＃xff0c;也叫万国码。

Python中&＃xff0c;多字节字符与数字间的转换&＃xff1a;

1. # Python3中&＃xff0c;字符对象&＃xff08;str), 可以使用 .encode方法将字符转为bytes对象

2. >>> "中国".encode("utf8")

3. b&＃39;\xe4\xb8\xad\xe5\x9b\xbd&＃39;

4. >>> "中国".encode("gbk")

5. b&＃39;\xd6\xd0\xb9\xfa&＃39;

7. # bytes对象转成字符

8. b&＃39;\xe4\xb8\xad\xe5\x9b\xbd&＃39;.decode("utf8")

9. &＃39;中国&＃39;

10. bytes([0xe4,0xb8,0xad,0xe5,0x9b,0xbd]).decode("utf8")

11. &＃39;中国&＃39;

使用hexdump查看文本文件的字符编码&＃xff1a;

1. $ file /tmp/python_chr.txt

2. /tmp/python_chr.txt: UTF-8 Unicode text

3. $ cat /tmp/python_chr.txt

4. Python

5. 中国

6. $ hexdump -C /tmp/python_chr.txt

7. 00000000 50 79 74 68 6f 6e 0a e4 b8 ad e5 9b bd 0a |Python........|

8. 0000000e

使用python来验证编码&＃xff1a;

1. # 前三个字符

2. >>> chr(0x50),chr(0x79),chr(0x74)

3. (&＃39;P&＃39;, &＃39;y&＃39;, &＃39;t&＃39;)

4. # 剩下的字符大家动手试一试&＃xff0c; 特别是汉字"中国"的编码

Python二进制相关

1、bytes对象

bytes是Python3中新增的一个处理二进制"流"的对象。可以下几种方式我们可以得到bytes对象&＃xff1a;

• 字符对象的encode方法
  

• 二进制文件read方法
  

• 网络socket的recv方法
  

• 使用b打头的字符申明
  

• 使用bytes对象初始化

一些简单的例子&＃xff1a;

1. >>> b&＃39;a&＃39;

2. b&＃39;a&＃39;

3. >>> type(b&＃39;a&＃39;)

5. >>> bytes([97])

6. b&＃39;a&＃39;

7. >>> bytes("中国",&＃39;utf8&＃39;)

8. b&＃39;\xe4\xb8\xad\xe5\x9b\xbd&＃39;

可以把bytes看作是一个特殊的数组&＃xff0c;由连续的字节(byte)组成&＃xff0c;单字节最大数不能超过255&＃xff0c;具有数组的切片&＃xff0c;迭代等特性&＃xff0c;它总是尝试以ASCII编码将数据转成可显示字符&＃xff0c;超出ASCII可显示范围则使用\x打头的二位十六进制进行显示。

bytes对象的本质是存的二进制数组&＃xff0c;存放的是0-255的数字数组&＃xff0c;它只有结合"字符集"才能转换正确的字符&＃xff0c;或者要结合某种"协议"才能解读出具体的"含义"&＃xff0c;这一点后面就会详细的讲到。

再来一个例子, 打印GBK编码表&＃xff1a;

1. # GBK编码从0x8140 开始&＃xff0c;显示 30 行

2. for row in [0x8140 &＃43; x*16 for x in range(30)]:

3. print(hex(row), end&＃61;" ")

4. # 每行显示16个

5. for i in range(16):

6. high &＃61; row&＃43;i >> 8 & 0xff # 高位

7. low &＃61; row&＃43;i & 0xff # 低位

8. try:

9. # 用bytes对象转换成GBK字符

10. print(bytes([high, low]).decode("gbk"), end&＃61;"")

11. except:

12. print(end&＃61;" ")

13. print("")

输出&＃xff1a;

1. 0x8140 丂丄丅丆丏丒丗丟丠両丣並丩丮丯丱

2. 0x8150 丳丵丷丼乀乁乂乄乆乊乑乕乗乚乛乢

3. 0x8160 乣乤乥乧乨乪乫乬乭乮乯乲乴乵乶乷

4. 0x8170 乸乹乺乻乼乽乿亀亁亂亃亄亅亇亊

5. 0x8180 亐亖亗亙亜亝亞亣亪亯亰亱亴亶亷亸

6. 0x8190 亹亼亽亾仈仌仏仐仒仚仛仜仠仢仦仧

7. 0x81a0 仩仭仮仯仱仴仸仹仺仼仾伀伂伃伄伅

8. 0x81b0 伆伇伈伋伌伒伓伔伕伖伜伝伡伣伨伩

9. 0x81c0 伬伭伮伱伳伵伷伹伻伾伿佀佁佂佄佅

10. 0x81d0 佇佈佉佊佋佌佒佔佖佡佢佦佨佪佫佭

11. 0x81e0 佮佱佲併佷佸佹佺佽侀侁侂侅來侇侊

12. 0x81f0 侌侎侐侒侓侕侖侘侙侚侜侞侟価侢

2、struct

计算机中几乎所有的数据都可以最终抽象成数字和字符来表示&＃xff0c;在C语言中用struct(结构体)来描述一个复杂的对象&＃xff0c;通过这个结构可以方便的将复杂对象转换成二进制流用于存储与网络传输。Python中提供了struct模块方便处理二进制流(bytes对象&＃xff09;与数字&＃xff0c;字符对象的转换功能。

3、用struct处理数字

1. >>> import struct

2. # 单字节数字

3. >>> struct.pack("

4. b&＃39;\xff&＃39;

6. # 双字节数字

7. >>> struct.pack("

8. b&＃39;\xff\x00&＃39;

10. # 四字节数字

11. >>> struct.pack("

12. b&＃39;\xff\x00\x00\x00&＃39;

14. # 八字节数字

15. >>> struct.pack("

16. b&＃39;\xff\x00\x00\x00\x00\x00\x00\x00&＃39;

18. #unpack可以找出8&＃xff0c;4&＃xff0c;2位符号整型的最大值

19. >>> struct.unpack("

20. (18446744073709551615,)

21. >>> struct.unpack("

22. (4294967295,)

23. >>> struct.unpack("

24. (65535,)

struct处理数字的要点有&＃xff1a;

• 字节数

• 有无符号位


• 字节序&＃xff0c;本文中均使用低字节在前的字节序"<"

4、用struct处理字符串

字符转换为bytes:

1. # 变长字符串&＃xff0c;以0xff结束

2. >>> struct.pack("<4s",b"cat")

3. b&＃39;cat\x00&＃39;

5. >>> struct.pack("<5s","中国".encode("gbk"))

6. b&＃39;\xd6\xd0\xb9\xfa\x00&＃39;

8. >>> struct.pack("<7s","中国".encode("utf8"))

9. b&＃39;\xe4\xb8\xad\xe5\x9b\xbd\x00&＃39;

11. # 定长字符串&＃xff0c;第1个字节为字符串的长度

12. >>> struct.pack("<4p",b"cat")

13. b&＃39;\x03cat&＃39;

15. >>> struct.pack("<5p","中国".encode("gbk"))

16. b&＃39;\x04\xd6\xd0\xb9\xfa&＃39;

18. >>> struct.pack("<7p","中国".encode("utf8"))

19. b&＃39;\x06\xe4\xb8\xad\xe5\x9b\xbd&＃39;

bytes转换为字符:

1. # 仅取一例&＃xff0c;其他的请自己动手试一试

2. >>> struct.unpack("<7p", b&＃39;\x06\xe4\xb8\xad\xe5\x9b\xbd&＃39;)[0].decode("utf8")

3. &＃39;中国&＃39;

需要特别说明的是&＃xff0c;unpack返回的是元组&＃xff0c;哪怕是只有一个元素&＃xff0c;这样做的好处是&＃xff0c;我们可以按照规则将多个数据的format写在一起&＃xff0c;让代码更加简洁&＃xff1a;

1. >>> struct.pack("

2. b&＃39;\x01\x00\x13\xea\x0c\x00\x00\x05alvin&＃39;

4. >>> struct.unpack("

5. (1, 19, 3306, b&＃39;alvin&＃39;)

6. >>> id, no, port, name &＃61; struct.unpack("

8. >>> id, no, port, name

9. (1, 19, 3306, b&＃39;alvin&＃39;)

这种写法会大量应用到后继的demo代码中&＃xff0c;请务必多加练习&＃xff0c;并仔细阅读官方文档。

Python Socket编程

简单说Socket编程&＃xff0c;就是面向网络传输层的接口编程&＃xff0c;系统通过IP地址和端口号建立起两台电脑之间网络连接&＃xff0c;并提供两个最基础的通信接口发送数据和接收数据&＃xff0c;供开发者调用&＃xff0c;先来看一个最简单的客户端Socket例子&＃xff1a;

1. import socket

3. # 创建一个socket对象

4. s &＃61; socket.socket(socket.AF_INET, socket.SOCK_STREAM)

5. # 建立连接

6. s.connect(("192.168.1.101", 3306))

7. # 接收数据

8. buf &＃61; s.recv(10240)

9. print(type(buf)) #

10. # 发送数据

11. s.send(b&＃39;hello&＃39;)

可以看出通过socket接收和发送的数据都是前面讲的bytes对象&＃xff0c;因为bytes对象本身只是一个二进制流&＃xff0c;所以在没有"协议"的前提下&＃xff0c;我们是无法理解传输内容的具体含义。常见的http,https,ftp,smtp,ssh协议都是建立socket通信之上的协议。换句说&＃xff0c;就是通socket编程可以实现与现有的任何协议进行通信。如果你熟悉了ssh协议&＃xff0c;那么实现ssh端口扫描程序就易如反掌了。

用socket不仅可以和其它协议的服务端进行通信&＃xff0c;而且可以实现socket服务端&＃xff0c;监听和处理来自client的连接和数据。

1. import socket

3. # 创建一个socket对象

4. s &＃61; socket.socket()

5. # 监听端口

6. s.bind((&＃39;127.0.0.1&＃39;, 8000))

7. s.listen(5)

9. while True:

10. conn, addr &＃61; s.accept()

11. conn.send(bytes(&＃39;Welcome python socket server.&＃39;, &＃39;utf8&＃39;))

12. # 关闭链接

13. conn.close()

通过上面两个简单的例子&＃xff0c;相信大家对Python的socket编程已经有一个初步的认识&＃xff0c;那就是"相当的简单"&＃xff0c;没有想象中那么复杂。

接下再来看一个多线程版的SocketServer, 可以通过telnet来实现一个网络计算器&＃xff1a;

1. # learn_socket3_server_mulit_thread.py

3. import threading

4. import socketserver

6. class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):

8. def handle(self):

9. """

10. 网络计算器&＃xff0c;返回表达式的值

11. """

12. while True:

13. try:

14. # 接收表达式数据

15. data &＃61; str(self.request.recv(1024), &＃39;ascii&＃39;).strip()

16. if "q" in data:

17. self.finish()

18. break

19. # 计算结果

20. response &＃61; bytes("{} &＃61; {}\r\n".format(data, eval(data)), &＃39;ascii&＃39;)

21. print(response.decode("ascii").strip())

22. # 返回结果

23. self.request.sendall(response)

24. except:

25. self.request.sendall(bytes("\n", &＃39;ascii&＃39;))

27. class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):

28. pass

30. if __name__ &＃61;&＃61; "__main__":

32. server &＃61; ThreadedTCPServer(("127.0.0.1", 9000), ThreadedTCPRequestHandler)

33. ip, port &＃61; server.server_address

35. server_thread &＃61; threading.Thread(target&＃61;server.serve_forever)

36. print(f"Calculator Server start at {ip} : {port}")

37. server_thread.start()

使用telnet进行测试&＃xff1a;

1. $ telnet 127.0.0.1 9000

2. Trying 127.0.0.1...

3. Connected to localhost.

4. Escape character is &＃39;^]&＃39;.

5. 12345679*81 # 按回车

6. 12345679*81 &＃61; 999999999 # 返回结果

7. 3&＃43;2-5*0 # Enter

8. 3&＃43;2-5*0 &＃61; 5

9. (123&＃43;123)*123 # Enter

10. (123&＃43;123)*123 &＃61; 30258

11. quit # Enter

服务端日志&＃xff1a;

1. Calculator Server start at 127.0.0.1 : 9000

2. 12345679*81 &＃61; 999999999

3. 3&＃43;2-5*0 &＃61; 5

4. (123&＃43;123)*123 &＃61; 30258

小结

理解二进制&＃xff0c;字符/编码&＃xff0c;socket通信&＃xff0c;以及如何使用Python来处理它们&＃xff0c;是实现BinlogServer最重要的基础&＃xff0c;由于篇幅问题&＃xff0c;很多知识点只能点到为止&＃xff0c;虽然很基础&＃xff0c;但是还是需要自己的动手去实验&＃xff0c;举一反三地多实践自己的想法&＃xff0c;会对理解后面的文章大有帮助。

只有会认真看文档的DBA才是好DBA&＃xff0c;只会认真看代码的Engineer,一定不是好Engineer。代码一定要运行起来&＃xff0c;On Runtime才会有价值&＃xff0c;才会让你变成好Engineer. ^_^

最后&＃xff0c;祝你编码快乐〜

相关文档

https://docs.python.org/3/library/struct.html

https://docs.python.org/3/library/socketserver.html

附&＃xff1a;基于mysqlbinlog命令的BinlogServer简单实现

1. #!/bin/sh

3. REMOTE_HOST&＃61;{{host}}

4. REMOTE_PORT&＃61;{{mysql_port}}

5. REMOTE_USER&＃61;{{mysql_repl_user}}

6. REMOTE_PASS&＃61;{{mysql_repl_password}}

8. BACKUP_BIN&＃61;/usr/local/mysql/bin/mysqlbinlog

9. LOCAL_BACKUP_DIR&＃61;/data/backup/mysql/binlog_3306

10. BACKUP_LOG&＃61;/data/backup/mysql/binlog_3306/backup_3306.log

12. FIRST_BINLOG&＃61;mysql-bin.000001

13. #time to wait before reconnecting after failure

14. SLEEP_SECONDS&＃61;10

16. ##create local_backup_dir if necessary

17. mkdir -p ${LOCAL_BACKUP_DIR}

19. cd ${LOCAL_BACKUP_DIR}

20. ## Function while loop &＃xff0c; After the connection is disconnected, wait for the specified time. &＃xff0c; Reconnect

21. while :

22. do

23. if [ &＃96;ls -A "${LOCAL_BACKUP_DIR}" |wc -l&＃96; -eq 0 ];then

24. LAST_FILE&＃61;${FIRST_BINLOG}

25. else

26. LAST_FILE&＃61;&＃96;ls -l ${LOCAL_BACKUP_DIR} | grep -v backuplog |tail -n 1 |awk &＃39;{print $9}&＃39;&＃96;

27. fi

29. ${BACKUP_BIN} --raw --read-from-remote-server --stop-never --host&＃61;${REMOTE_HOST} --port&＃61;${REMOTE_PORT} --user&＃61;${REMOTE_USER} --password&＃61;${REMOTE_PASS} ${LAST_FILE}

30. echo "&＃96;date &＃43;"%Y/%m/%d %H:%M:%S"&＃96; mysqlbinlog Stop it &＃xff0c; Return code &＃xff1a;$?" | tee -a ${BACKUP_LOG}

31. echo "${SLEEP_SECONDS} After the second connect and continue to backup " | tee -a ${BACKUP_LOG}

32. sleep ${SLEEP_SECONDS}

33. done

扫码加入MySQL技术Q群

&＃xff08;群号&＃xff1a;650149401&＃xff09;

   

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