Q4M使用手册

 

一、概述

1. Q4M模型

是基于MySQL存储引擎的轻量级消息队列，通过扩展SQL语法来操作消息队列，使用简单，容易上手，开发人员基本不用再进行学习和熟悉。

Q4M支持多发送方，多接收方，接收方相互不影响，支持可靠获取，支持多队列，宕机后数据可恢复，可用SQL方便查看队列状态，实测单队列单线程的入队性能上限为7405QPS (单消息10b)，出队列性能上限在9375QPS，单队列50线程的入队列性能上限为81632QPS，出队列性能上限为62500QPS。

2. 属主状态（OWNER MODE）说明

为了保证可靠获取，Q4M引入了属主状态（OWNER MODE）和无属主状态（NON-OWNER MODE）概念，此为Q4M最大的特性。当消息出队列后，则该消息先进入属主状态（OWNER MODE），即该消息只对属主所在线程可见，对其它线程不可见。

通过调用queue_wait()取消息出队并进入属主状态，每次获取消息最多不超过一条(队列为空返回零条)，调用queue_end()结束属主状态，并删除已出队消息；调用queue_abort()，放弃属主状态，已出队消息重新回收到队列中；若在属主状态中连接中断，则已出队消息重新回收到队列中；若连续调用queue_wait()，则会自动删除上次出队消息并获取下一条信息，即可理解为在第两次queue_wait()前自动执行了queue_end()。详细可见下文示例。

3. 使用限制及注意点

Ø 没有主键或索引支持；

Ø 不支持AUTO_INCREMENT自增列，默认按插入顺序排序；

Ø 不支持update，可支持insert和delete；

Ø 属主状态（owner mode）的消息获取超时时间默认为60秒，注意此为获取时间，不是消息使用时间。如队列为空时，queue_wait()会等待直至超时后返回；

Ø 若连接异常断开，消息自动回收到队列中，因此建议将MYSQL 参数interactive_timeout、 wait_timeout调短到10分钟以内，连接超时断开后可自动回收消息；

Ø 一条消息上限为512MB，表上限为2^63 bytes，即1048576 TB；

Ø 内存不足时可能会导致MySQL Crash；

Ø 若压缩数据时宕机，内存中的消息可能会丢失；

Ø 不支持主从复制，可通过搭建多套Q4M进行容灾

二．使用示例

1. 创建队列

一个表就是一个队列，每一行记录是一条消息。若需创建一队列，只需建一queue引擎的table即可，若需要多队列，则创建多个表即可。

MySQL> CREATE TABLE my_queue (v1 int not null, v2 varchar(255)) ENGINE=queue;

2. 消息入队

只需用正常的sql语法insert记录即可将消息添加到队列。

MySQL> 

INSERT INTO my_queue (v1, v2) VALUES (1, "Tian");

INSERT INTO my_queue (v1, v2) VALUES (2, "Jing");

INSERT INTO my_queue (v1, v2) VALUES (3, "1983");

INSERT INTO my_queue (v1, v2) VALUES (4, "IN");

INSERT INTO my_queue (v1, v2) VALUES (5, "GANJI");

INSERT INTO my_queue (v1, v2) VALUES (6, "HAPPY");

MySQL> select * from my_queue; 

+----+-------+

| v1 | v2    |

+----+-------+

|  1 | Tian    |

|  2 | Jing   |

|  3 | 1983  |

|  4 | IN    |

|  5 | GANJI |

|  6 | HAPPY |

+----+-------+

6 rows in set (0.00 sec)

3. 消息出队

调用queue_wait()使消息出队，获取的消息进入属主状态(可见1.2解释)，在本线程内，只能查到已出队消息。

MySQL>SELECT * FROM my_queue WHERE queue_wait('my_queue');

+----+------+

| v1 | v2   |

+----+------+

|  1 | Tian   |

+----+------+

1 row in set (0.00 sec)

MySQL>SELECT * FROM my_queue;

+----+------+

| v1 | v2   |

+----+------+

|  1 | Tian   |

+----+------+

1 row in set (0.00 sec)

调用queue_end()结束属主状态，自动删除上次出队的消息。这时select * 可查到队列中所有消息了。

MySQL>select queue_end(); 

+-------------+

| queue_end() |

+-------------+

|           1 |

+-------------+

1 row in set (0.00 sec)

MySQL>SELECT * FROM my_queue;

+----+-------+

| v1 | v2    |

+----+-------+

|  2 | Jing   |

|  3 | 1983  |

|  4 | IN    |

|  5 | GANJI |

|  6 | HAPPY |

+----+-------+

5 rows in set (0.00 sec)

若调用queue_abort()则放弃属主状态，已出队消息自动回收到原队列中。

MySQL>SELECT * FROM my_queue WHERE queue_wait('my_queue');

+----+------+

| v1 | v2   |

+----+------+

|  1 | Tian   |

+----+------+

1 row in set (0.00 sec)

MySQL>SELECT * FROM my_queue ; 

+----+------+

| v1 | v2   |

+----+------+

|  1 | Tian   |

+----+------+

1 row in set (0.00 sec)

MySQL>select queue_abort();                               

+---------------+

| queue_abort() |

+---------------+

|             1 |

+---------------+

1 row in set (0.00 sec)

MySQL>SELECT * FROM my_queue ; 

+----+-------+

| v1 | v2    |

+----+-------+

|  1 | Tian    |   # queue_abort()该消息重新回收到队列中

|  2 | Jing  |

|  3 | 1983  |

|  4 | IN    |

|  5 | GANJI |

|  6 | HAPPY |

+----+-------+

6 rows in set (0.00 sec)

若连续调用queue_wait()，则会自动删除上次出队消息并获取下一条信息，即可理解为在第两次queue_wait()前自动执行了queue_end()

MySQL>SELECT * FROM my_queue WHERE queue_wait('my_queue');

+----+------+

| v1 | v2   |

+----+------+

|  1 | Tian   |

+----+------+

1 row in set (0.00 sec)

MySQL>SELECT * FROM my_queue WHERE queue_wait('my_queue');

+----+------+

| v1 | v2   |

+----+------+

|  2 | Jing  |

+----+------+

1 row in set (0.02 sec)

MySQL>SELECT * FROM my_queue WHERE queue_wait('my_queue');

+----+------+

| v1 | v2   |

+----+------+

|  3 | 1983 |

+----+------+

1 row in set (0.01 sec)

MySQL>select queue_end();    

+-------------+

| queue_end() |

+-------------+

|           1 |

+-------------+

1 row in set (0.00 sec)

MySQL>SELECT * FROM my_queue ;  #前三条消息都已被取出

+----+-------+

| v1 | v2    |

+----+-------+

|  4 | IN    |

|  5 | GANJI |

|  6 | HAPPY |

+----+-------+

3 rows in set (0.00 sec)

4. 按条件进行消息出队

Q4M也支持按某条件获取消息并出队，使用方法为queue_wait('table:col>N')：

MySQL>SELECT * FROM my_queue ;                          

+----+-------+

| v1 | v2    |

+----+-------+

|  1 | Tian    |

|  2 | Jing  |

|  3 | 1983  |

|  4 | IN    |

|  5 | GANJI |

|  6 | HAPPY |

+----+-------+

6 rows in set (0.00 sec)

MySQL>SELECT * FROM my_queue WHERE queue_wait('my_queue:v1>=5');

+----+-------+

| v1 | v2    |

+----+-------+

|  5 | GANJI |    #取出的值为v1>=5的第一条消息

+----+-------+

1 row in set (0.01 sec)

目前Q4M支持如下运算符：

- ~ ()

* div % mod

+ -

<< >>

&

|

= != <= < >= >

not

&& and

xor

|| or

5. 消息出队的伪代码实现

如下是消息出队的伪代码实现：

while (true) {

   rows := SELECT * FROM my_queue

     WHERE queue_wait('my_queue');     # 消息出队并进入属主状态

  if (count(rows) != 0)                # 如果有消息，则

    handle_row(rows[0]);               # 进行消息处理

  SELECT queue_end();                  # 结束属主状态

}