memcacheQ

阎自怡

2023-12-01

memcacheQ是一个单纯的分布式消息队列服务。它的安装依赖于BerkeleyDB 和 libevent，所以要先安装这BerkeleyDB和libevent：

Berkeley DB 4.7 or later

Download from <http://www.oracle.com/database/berkeley-db/db/index.html>

How to install BerkekeyDB:

$tar xvzf db-4.7.25.tar.gz
$cd db-4.7.25/
$cd build_unix/
$../dist/configure
$make
$sudo make install

libevent 1.4.x or later

Download from <http://monkey.org/~provos/libevent/>

How to install libevent:

$tar xvzf libevent-1.4.x-stable.tar.gz
$cd libevent-1.4.x-stable
$./configure
$make
$sudo make install

On a linux, load .so file by add two line in /etc/ld.so.conf:

/usr/local/lib
/usr/local/BerkeleyDB.4.7/lib

Then, run 'ldconfig'.

Building MemcacheQ

On a *nix, just following:

$tar xvzf memcacheq-0.2.x.tar.gz
$cd memcacheq-0.2.x
$./configure --enable-threads
$make
$sudo make install

启动memcacheQ

使用memcacheq -h 的命令来查看命令行选项
启动memcacheq：memcacheq -d -r -u root -H /data1/memcacheq -N -R -v -L 1024 -B 1024 > /data1/mq_error.log 2>&1

Commands

Only two commands are used to operate the queue:

Append a message to the tail of queue:

set <queue name> <flags> 0 <message_len>\r\n
<put your message body here>\r\n
STORED\r\n

Note: MQ will create a new queue automatically if your queue is not existed. The original 'expire time' field is ignored by server.

Consume a message from the head of queue:

get <queue name>\r\n
VALUE <queue name> <flags> <message_len>\r\n
<your message body will come here>\r\n
END\r\n

Examples

Assuming you are using PHP memcache<http://www.php.net/memcache>:

<?php
/* connect to memcached server */
$memcache_obj = memcache_connect('memcacheq_host', 21201);

/* append a message to queue */
memcache_set($memcache_obj, 'demoqueue1', 'message body here', 0, 0);

/* consume a message from 'demoqueue1' */
memcache_get($memcache_obj, 'demoqueue1');

memcache_close($memcache_obj);
?>

使用以上命令启动mq后，（注意上面的-B参数表示messag的body长度不能超过1024 bytes），使用mq时只需要用到两个命令：set和get：

set <queue name> <flags> 0 <message_len>\r\n
<put your message body here>\r\n
STORED\r\n

get <queue name>\r\n
VALUE <queue name> <flags> <message_len>\r\n
<your message body will come here>\r\n
END\r\n

可以看到，和memcache协议基本一致，只是把key name换成queue name，而且在set的命令中，忽略了expire_time的参数。毕竟mq的数据存储是存在berkeleyDB中，做了持久化存储，没有内存的过期时间。

当使用set命令时，就向指定的消息队列中写入了一条新消息，也就是向BerkeleyDB中新insert了一条数据，当使用get命令时，就从指定队列中取出一条新消息，也就是向BerkeleyDB中delete了一条数据。当使用stats查看一个指定队列时，可以看到这个队列一共接收了多少消息，其中被取出了多少条。

示例：

fengbo@onlinegame-10-121:~$ telnet 127.0.0.1 22202

Trying 127.0.0.1…

Connected to 127.0.0.1.

Escape character is ‘^]’.

set q4 0 0 5

hello

STORED

set q4 0 0 5

world

STORED

stats queue

STAT q4 2/0

END

get q4

VALUE q4 0 5

hello

END

stats queue

STAT q4 2/1

END

上面执行了两次set的命令，使用stats queue查看时，可以看到q4的队列中共有消息2条，已取出0条；当使用get取出第一条后，再此使用stats queue查看，q4中消息有2条，其中已取出1条。

Other tips

use 'stats queue' to see your current queues:

$ telnet 127.0.0.1 22201
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
stats queue
STAT test1
STAT test2
STAT test3
STAT test4
END

delete a queue:

$ telnet 127.0.0.1 22201
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
delete test1
DELETED

'db_stat' a queue to see how many records now in:

$ cd <your queue dir>
$ /usr/local/BerkeleyDB.4.7/bin/db_stat -d test1
Mon Sep 22 20:25:56 2008      Local time
42253   Queue magic number
4       Queue version number
1024    Fixed-length record size
0x20    Fixed-length record pad
4096    Underlying database page size
131072  Underlying database extent size
100000  Number of records in the database
33334   Number of database pages
2048    Number of bytes free in database pages (99% ff)
1       First undeleted record
100001  Next available record number