分布式数据库中间件–(3)Cobar对简单select命令的处理过程

在认证成功后Cobar会将该连接的回调处理函数由FrontendAuthenticator（前端认证处理器）设置成FrontendCommandHanler（前端命令处理器）。 所以在客户端再次向Cobar发送请求报文的时候，前端命令处理器会处理该连接。下面详细分析一下简单select语句的执行过

在认证成功后Cobar会将该连接的回调处理函数由FrontendAuthenticator（前端认证处理器）设置成FrontendCommandHanler（前端命令处理器）。

所以在客户端再次向Cobar发送请求报文的时候，前端命令处理器会处理该连接。下面详细分析一下简单select语句的执行过程。

1、事件的产生

NIOReactor的R线程一直在监听selector上的每个连接的感兴趣事件是否发生，当客户端发送了一条select * from tb1，select函数会返回，然后获取到该连接SelectionKey，并且该SelectKey的兴趣事件是OP_READ。此时会调用read(NIOConnection)函数。

01	public void run() {

02	final Selector selector = this.selector;

03	for (;;) {

04	++reactCount;

05

try {

06	int res = selector.select();

07	LOGGER.debug(reactCount + ">>NIOReactor接受连接数:" + res);

08	register(selector);

09	Set keys = selector.selectedKeys();

10

try {

11	for (SelectionKey key : keys) {

12	Object att = key.attachment();

13	if (att != null && key.isValid()) {

14	int readyOps = key.readyOps();

15	if ((readyOps & SelectionKey.OP_READ) != 0) {

16	LOGGER.debug("select读事件");

17	read((NIOConnection) att);

18	..............................

19

}

20	...........................

21

}

22	} ..................

23	} ............

24

}

25

}

2、调用该连接的read函数进行处理

该函数在上一篇中提到过，该函数的实现在AbstractConnection中，实现从channel中读取数据到缓冲区，然后从缓冲区完整的取出整包数据交给FrontendConnection类的handle()函数处理。

该函数交给processor进行异步处理。从processor中的线程池获取一个线程来执行该任务。这里调用具体的handler来进行处理。

刚开始提到的，当认证成功后，Cobar将连接的回调处理函数设置为FrontendCommandHandler。所以这里会调用前端命令处理器的handler函数进行数据的处理。

在这里需要先了解MySQL数据包的格式：

MySQL客户端命令请求报文

该处理函数如下：

01	public void handle(byte[] data) {

02	LOGGER.info("data[4]:"+data[4]);

03	switch (data[4]) {

04	case MySQLPacket.COM_INIT_DB:

05	commands.doInitDB();

06	source.initDB(data);

07

break;

08	case MySQLPacket.COM_QUERY:

09	commands.doQuery();

10	source.query(data);

11

break;

12	case MySQLPacket.COM_PING:

13	commands.doPing();

14	source.ping();

15

break;

16	case MySQLPacket.COM_QUIT:

17	commands.doQuit();

18	source.close();

19

break;

20	case MySQLPacket.COM_PROCESS_KILL:

21	commands.doKill();

22	source.kill(data);

23

break;

24	case MySQLPacket.COM_STMT_PREPARE:

25	commands.doStmtPrepare();

26	source.stmtPrepare(data);

27

break;

28	case MySQLPacket.COM_STMT_EXECUTE:

29	commands.doStmtExecute();

30	source.stmtExecute(data);

31

break;

32	case MySQLPacket.COM_STMT_CLOSE:

33	commands.doStmtClose();

34	source.stmtClose(data);

35

break;

36	case MySQLPacket.COM_HEARTBEAT:

37	commands.doHeartbeat();

38	source.heartbeat(data);

39

break;

40

default:

41	commands.doOther();

42	source.writeErrMessage(ErrorCode.ER_UNKNOWN_COM_ERROR, "Unknown command");

43

}

44

}

由于每个报文都有消息头，消息头固定的是4个字节，前3个字节是消息长度，后面的一个字节是报文序号，如下所示

所以data[4]是第五个字节。也就是消息体的第一个字节。客户端向Cobar端发送的是命令报文，第一个字节是具体的命令。

如果是select语句，那么data[4]就是COM_QUERY，然后会调用具体连接的query成员函数，其定义在FrontendConnection类中。

01	public void query(byte[] data) {

02	if (queryHandler != null) {

03

// 取得语句

04	MySQLMessage mm = new MySQLMessage(data);

05	mm.position(5);

06	String sql = null;

07

try {

08	sql = mm.readString(charset);

09	} catch (UnsupportedEncodingException e) {

10	writeErrMessage(ErrorCode.ER_UNKNOWN_CHARACTER_SET, "Unknown charset '" + charset + "'");

11

return;

12

}

13	if (sql == null || sql.length() == 0) {

14	writeErrMessage(ErrorCode.ER_NOT_ALLOWED_COMMAND, "Empty SQL");

15

return;

16

}

17	LOGGER.debug("解析的SQL语句:"+sql);

18

// 执行查询

19	queryHandler.query(sql);

20

} else {

21	writeErrMessage(ErrorCode.ER_UNKNOWN_COM_ERROR, "Query unsupported!");

22

}

23

}

首先新建一个MySQLMessage对象，将数据包的索引位置定位到第6个字节位置处。然后将后面的所有的字节读取成指定编码格式的SQL语句，这里就形成了完整的SQL语句。

查询的时候Cobar控制台输出如下内容：

11:35:33,392 INFO data[4]:3

解析出SQL语句后交给queryHandler处理。该对象是在新建连接的时候设置的ServerQueryHandler类，其实现的query函数如下：

01	public void query(String sql) {

02	//这里就得到了完整的SQL语句，接收自客户端

03	ServerConnection c = this.source;

04	if (LOGGER.isDebugEnabled()) {

05	LOGGER.debug(new StringBuilder().append(c).append(sql).toString());

06

}

07	//该函数对SQL语句的语法和语义进行分析，并返回SQL语句的对于类型，执行相应的操作

08	int rs = ServerParse.parse(sql);

09	switch (rs & 0xff) {

10	.......................

11	case ServerParse.SELECT:

12	//select操作执行

13	SelectHandler.handle(sql, c, rs >>> 8);

14

break;

15	.......................

16

}

17

}

首先对SQL语句进程解析，通过parse函数对语句解析后返回语句类型的编号。

如果语句没有语法错误，则直接交给SelectHandler进行处理。如果是一般的select语句，则直接调用ServerConnection的execute执行sql

c.execute(stmt, ServerParse.SELECT);

在ServerConnection中的execute函数中需要进行路由检查，因为select的数据不一定在一个数据库中，需要按拆分的规则进行路由的检查。

1

// 路由计算

2	RouteResultset rrs = null;

3

try {

4	rrs = ServerRouter.route(schema, sql, this.charset, this);

5	LOGGER.debug("路由计算结果:"+rrs.toString());

6

}

具体的路由算法也是比较复杂，以后会专门分析。

Cobar的DEBUG控制台输出路由的计算结果如下：

11:35:33,392 DEBUG 路由计算结果:select * from tb2, route={

该条SQL语句的select内容分布在dnTset2和dnTest3中，所以要分别向这两个数据库进行查询。

经过比较复杂的资源处理最后在每个后端数据库上执行函数execute0。

01	private void execute0(RouteResultsetNode rrn, Channel c, boolean autocommit, BlockingSession ss, int flag) {

02	ServerConnection sc = ss.getSource();

03	.........................

04

try {

05	// 执行并等待返回

06	BinaryPacket bin = ((MySQLChannel) c).execute(rrn, sc, autocommit);

07	// 接收和处理数据，执行到这里就说明上面的执行已经得到执行结果的返回

08	final ReentrantLock lock = MultiNodeExecutor.this.lock;

09	lock.lock();

10

try {

11	switch (bin.data[0]) {

12	case ErrorPacket.FIELD_COUNT:

13	c.setRunning(false);

14	handleFailure(ss, rrn, new BinaryErrInfo((MySQLChannel) c, bin, sc, rrn));

15

break;

16	case OkPacket.FIELD_COUNT:

17	OkPacket ok = new OkPacket();

18	ok.read(bin);

19	affectedRows += ok.affectedRows;

20	// set lastInsertId

21	if (ok.insertId > 0) {

22	insertId = (insertId == 0) ? ok.insertId : Math.min(insertId, ok.insertId);

23

}

24	c.setRunning(false);

25	handleSuccessOK(ss, rrn, autocommit, ok);

26

break;

27	default: // HEADER|FIELDS|FIELD_EOF|ROWS|LAST_EOF

28	final MySQLChannel mc = (MySQLChannel) c;

29	if (fieldEOF) {

30	for (;;) {

31	bin = mc.receive();

32	switch (bin.data[0]) {

33	case ErrorPacket.FIELD_COUNT:

34	c.setRunning(false);

35	handleFailure(ss, rrn, new BinaryErrInfo(mc, bin, sc, rrn));

36

return;

37	case EOFPacket.FIELD_COUNT:

38	handleRowData(rrn, c, ss);

39

return;

40

default:

41

continue;

42

}

43

}

44

} else {

45	bin.packetId = ++packetId;// HEADER

46	List headerList = new LinkedList();

47	headerList.add(bin);

48	for (;;) {

49	bin = mc.receive();

50	switch (bin.data[0]) {

51	case ErrorPacket.FIELD_COUNT:

52	c.setRunning(false);

53	handleFailure(ss, rrn, new BinaryErrInfo(mc, bin, sc, rrn));

54

return;

55	case EOFPacket.FIELD_COUNT:

56	bin.packetId = ++packetId;// FIELD_EOF

57	for (MySQLPacket packet : headerList) {

58	buffer = packet.write(buffer, sc);

59

}

60	headerList = null;

61	buffer = bin.write(buffer, sc);

62	fieldEOF = true;

63	handleRowData(rrn, c, ss);

64

return;

65

default:

66	bin.packetId = ++packetId;// FIELDS

67	switch (flag) {

68	case RouteResultset.REWRITE_FIELD:

69	StringBuilder fieldName = new StringBuilder();

70	fieldName.append("Tables_in_").append(ss.getSource().getSchema());

71	FieldPacket field = PacketUtil.getField(bin, fieldName.toString());

72	headerList.add(field);

73

break;

74

default:

75	headerList.add(bin);

76

}

77

}

78

}

79

}

80

}

81	} finally {

82	lock.unlock();

83

}

84	}//异常处理....................

85

}

这里真正的执行SQL语句，然后等待后端执行语句的返回数据，在成功获取后端Mysql返回的结果后，该函数返回的数据包是结果集数据包。

当客户端发起认证请求或命令请求后，服务器会返回相应的执行结果给客户端。客户端在收到响应报文后，需要首先检查第1个字节的值，来区分响应报文的类型。

响应报文类型	第1个字节取值范围
OK 响应报文	0×00
Error 响应报文	0xFF
Result Set 报文	0×01 – 0xFA
Field 报文	0×01 – 0xFA
Row Data 报文	0×01 – 0xFA
EOF 报文	0xFE

注：响应报文的第1个字节在不同类型中含义不同，比如在OK报文中，该字节并没有实际意义，值恒为0×00；而在Result Set报文中，该字节又是长度编码的二进制数据结构（Length Coded Binary）中的第1字节。

Result Set 消息分为五部分，结构如下：

结构	说明
[Result Set Header]	列数量
[Field]	列信息（多个）
[EOF]	列结束
[Row Data]	行数据（多个）
[EOF]	数据结束

函数执行完成后，返回的结果都放入LinkedList中，当读取结果完成后放入多节点执行器的缓冲区。如果buffer满了，就通过前端连接写出给客户端。