php 频繁插库处理队列,安装配置MemcacheQ消息队列服务

MemcacheQ介绍

队列(Queue)是一种常用的数据结构。在队列这种数据结构中，最先插入的元素将会最先被取出；反之最后插入的元素将会最后被取出，因此队列又称为“先进先出”(FIFO：First In First Out)的线性表。

加入元素的一端叫“队尾”，取出元素的一端叫“队头”。利用消息队列可以很好地异步处理数据的传送和存储，当遇到频繁且密集地向后端数据库中插入数据时，就可采用消息队列来异步处理这些数据写入。

MemcacheQ是一款基于Memcache协议的开源消息队列服务软件，由于其遵循了Memcache协议，因此开发成本很低，不需要学习额外的知识便可轻松掌握。

我在最近的一个项目中也应用了MemcacheQ，下面我将分享一下MemcacheQ在Linux中的编译和安装过程。

首先，MemcacheQ依赖于BerkeleyDB和Libevent，如果服务器中曾经安装过Memcached，那么Libevent应该已经存在了，否则就需要先下载安装Libevent。

下载链接如下：

MemcacheQ特点

1.简单高效，基于memcache协议，这意味着只要客户端支持memcache协议即可使用。

2.队列数据存储于BDB,持久保存。

3.并发性能好。

4.支持多条队列。

memcacheQ依赖于Berkeley DB和libevent。Berkeley DB用于持久化存储队列的数据，避免在memcacheq崩溃或这服务器当掉时候，不至于数据丢失。对于并发量较高的web环境，特别是数据库写入操作过多的情景，使用队列可大大缓解因并发问题造成的数据库锁死问题。

memcacheQ最大的优势是：它是基于Memcached开发的，可以通过各种Memcached命令对它进行操作。基于Memcached开发的应用完全不需要做任何修改。

启动MemcacheQ

MemcacheQ常用参数

-p #TCP监听端口(default: 22201).

-U #UDP监听端口(default: 0,off).

-s #unix socket路径(不支持网络).

-a #unix socket访问掩码(default 0700).

-l #监听网卡.

-d #守护进程.

-r #最大化核心文件限制.

-u #以用户身份运行(only when run as root).

-c #最大并发连接数(default is 1024).

-v #详细输出 (print errors/warnings while in event loop).

-vv #更详细的输出 (also print client commands/reponses).

-i #打印许可证信息.

-P #PID文件.

-t #线程数(default 4).

1

2

3

4

5

6

7

8

9

10

11

12

13

14

-p#TCP监听端口(default: 22201).

-U#UDP监听端口(default: 0,off).

-s#unix socket路径(不支持网络).

-a#unix socket访问掩码(default 0700).

-l#监听网卡.

-d#守护进程.

-r#最大化核心文件限制.

-u#以用户身份运行(only when run as root).

-c#最大并发连接数(default is 1024).

-v#详细输出 (print errors/warnings while in event loop).

-vv#更详细的输出 (also print client commands/reponses).

-i#打印许可证信息.

-P#PID文件.

-t#线程数(default 4).

Berkerledb常用参数

-m #BerkeleyDB内存缓存大小,default is 64MB.

-A #底层页面大小,default is 4096, (512B ~ 64KB, power-of-two).

-H #数据库家目录,default is '/data1/memcacheq'.

-L #日志缓冲区大小,default is 32KB.

-C #多少秒checkpoint一次,0 for disable, default is 5 minutes.

-T #多少秒memp_trickle一次,0 for disable, default is 30 seconds.

-S #多少秒queue stats dump一次,0 for disable, default is 30 seconds.

-e #达到缓存百分之多少需要刷新,default is 60%.

-E #一个单一的DB文件有多少页,default is 16*1024, 0 for disable.

-B #指定消息体的长度,单位字节,default is 1024.

-D #多少毫秒做一次死锁检测(deadlock detecting),0 for disable, default is 100ms.

-N #开启DB_TXN_NOSYNC获得巨大的性能改善,default is off.

-R #自动删除不再需要的日志文件,default is off.

1

2

3

4

5

6

7

8

9

10

11

12

13

-m#BerkeleyDB内存缓存大小,default is 64MB.

-A#底层页面大小,default is 4096, (512B ~ 64KB, power-of-two).

-H#数据库家目录,default is '/data1/memcacheq'.

-L#日志缓冲区大小,default is 32KB.

-C#多少秒checkpoint一次,0 for disable, default is 5 minutes.

-T#多少秒memp_trickle一次,0 for disable, default is 30 seconds.

-S#多少秒queue stats dump一次,0 for disable, default is 30 seconds.

-e#达到缓存百分之多少需要刷新,default is 60%.

-E#一个单一的DB文件有多少页,default is 16*1024, 0 for disable.

-B#指定消息体的长度,单位字节,default is 1024.

-D#多少毫秒做一次死锁检测(deadlock detecting),0 for disable, default is 100ms.

-N#开启DB_TXN_NOSYNC获得巨大的性能改善,default is off.

-R#自动删除不再需要的日志文件,default is off.

创建数据目录

$ mkdir /data/memcacheq/11311/{db,log} -p

$ chown -R nobody.nobody /data/memcacheq/

1

2

$mkdir/data/memcacheq/11311/{db,log}-p

$chown-Rnobody.nobody/data/memcacheq/

启动memcacheq

$ /usr/local/memcacheq/bin/memcacheq -d -unobody -r -l 127.0.0.1 -p11311 -H /data/memcacheq/11311/db -N -R -v -L 1024 -B 1024 \

-P/data/memcacheq/11311/memcacheq.pid > /data/memcacheq/11311/log/mcq_error.log 2>&1

1

2

$/usr/local/memcacheq/bin/memcacheq-d-unobody-r-l127.0.0.1-p11311-H/data/memcacheq/11311/db-N-R-v-L1024-B1024\

-P/data/memcacheq/11311/memcacheq.pid>/data/memcacheq/11311/log/mcq_error.log2>&1

测试MemcacheQ

1.使用mcq时只需要用到两个命令：set和get

set 0

STORED

get

VALUE

END

1

2

3

4

5

6

7

set0

STORED

get

VALUE

END

测试

$ telnet 127.0.0.1 11311

set queue 0 0 3

abc

STORED

get queue

VALUE queue 0 3

abc

END

get queue_test

END

1

2

3

4

5

6

7

8

9

10

$telnet127.0.0.111311

setqueue003

abc

STORED

getqueue

VALUEqueue03

abc

END

getqueue_test

END

可以看到，和memcache协议基本一致，只是把key name换成queue name，而且在set的命令中，忽略了expire_time的参数。毕竟mq的数据存储是存在berkeleyDB中，做了持久化存储，没有内存的过期时间。当使用set命令时，就向指定的消息队列中写入了一条新消息，也就是向BerkeleyDB中新insert了一条数据，当使用get命令时，就从指定队列中取出一条新消息，也就是向BerkeleyDB中delete了一条数据。

2.查看统计情况

stats            #查看memcacheQ统计情况;

stats queue #查看队列统计情况;

1

2

stats      #查看memcacheQ统计情况;

statsqueue#查看队列统计情况;

3.删除一个队列

delete queue

1

deletequeue

4.也可以使用db_stat查看一个队列有多少记录

$ /usr/local/berkeleydb/bin/db_stat -d /data/memcacheq/11311/db/queue

Mon Nov 7 18:55:22 2016Local time

42253Queue magic number

4Queue version number

1024Fixed-length record size

0x20Fixed-length record pad

4096Underlying database page size

16384Underlying database extent size

0Number of records in the database

0Number of data items in the database

1Number of database pages

3072Number of bytes free in database pages (25% ff)

2First undeleted record

2Next available record number

1

2

3

4

5

6

7

8

9

10

11

12

13

14

$/usr/local/berkeleydb/bin/db_stat-d/data/memcacheq/11311/db/queue

MonNov718:55:222016Localtime

42253Queuemagicnumber

4Queueversionnumber

1024Fixed-lengthrecordsize

0x20Fixed-lengthrecordpad

4096Underlyingdatabasepagesize

16384Underlyingdatabaseextentsize

0Numberofrecordsinthedatabase

0Numberofdataitemsinthedatabase

1Numberofdatabasepages

3072Numberofbytesfreeindatabasepages(25%ff)

2Firstundeletedrecord

2Nextavailablerecordnumber

如果您觉得本站对你有帮助，那么可以支付宝扫码捐助以帮助本站更好地发展，在此谢过。