les*_*es2 7 python java message-queue rabbitmq nosql
从不太酷的问题类别......
"队列式的东西"我的意思是支持以下操作:
可选操作类似于:
如果可以以分布式方式对队列执行以下操作(多个客户端与队列交互),那将是理想的:
queue = ...
queue.append( a )
queue.append( b )
queue.append( c )
print queue
"a b c"
queue.promote( b.id )
print queue
"b a c"
queue.demote( a.id )
"b c a"
x = queue.take()
print x
"b"
print queue
"c a"
是否有任何特别适合此用例的数据存储？即使多个用户同时修改队列,队列也应始终处于一致状态.
如果不是为了促进/降级/移动要求,那就不会有太大的问题.
编辑:如果有Java和/或Python库来完成上述任务的加分点.
解决方案应该非常好地扩展
Redis支持列表和有序集:http://redis.io/topics/data-types#lists
它还支持事务和发布/订阅消息传递.所以,是的,我想说这可以在redis上轻松完成.
更新:事实上,其中大约80%已经多次完成:http://www.google.co.uk/search？q = python + redis +queue
其中一些命中可以升级,以添加你想要的.您必须使用事务来实现提升/降级操作.
可以在服务器端使用lua来创建该功能,而不是在客户端代码中使用它.或者,您可以在服务器上创建一个围绕redis的瘦包装器,它只实现您想要的操作.
我认为 python 和几个库足以解决这个问题，而不是寻找像 RabbitMQ、Redis 或 RDBMS 这样的数据存储。有些人可能会抱怨这种自己动手的方法是在重新发明轮子，但我更喜欢运行一百行 python 代码而不是管理另一个数据存储。
您定义的操作：追加、获取、提升和降级描述了优先级队列。不幸的是 python 没有内置的优先级队列数据类型。但它确实有一个名为 heapq 的堆库，并且优先级队列通常以堆的形式实现。这是我满足您要求的优先级队列的实现：
class PQueue:
"""
Implements a priority queue with append, take, promote, and demote
operations.
"""
def __init__(self):
"""
Initialize empty priority queue.
self.toll is max(priority) and max(rowid) in the queue
self.heap is the heap maintained for take command
self.rows is a mapping from rowid to items
self.pris is a mapping from priority to items
"""
self.toll = 0
self.heap = list()
self.rows = dict()
self.pris = dict()
def append(self, value):
"""
Append value to our priority queue.
The new value is added with lowest priority as an item. Items are
threeple lists consisting of [priority, rowid, value]. The rowid
is used by the promote/demote commands.
Returns the new rowid corresponding to the new item.
"""
self.toll += 1
item = [self.toll, self.toll, value]
self.heap.append(item)
self.rows[self.toll] = item
self.pris[self.toll] = item
return self.toll
def take(self):
"""
Take the highest priority item out of the queue.
Returns the value of the item.
"""
item = heapq.heappop(self.heap)
del self.pris[item[0]]
del self.rows[item[1]]
return item[2]
def promote(self, rowid):
"""
Promote an item in the queue.
The promoted item swaps position with the next highest item.
Returns the number of affected rows.
"""
if rowid not in self.rows: return 0
item = self.rows[rowid]
item_pri, item_row, item_val = item
next = item_pri - 1
if next in self.pris:
iota = self.pris[next]
iota_pri, iota_row, iota_val = iota
iota[1], iota[2] = item_row, item_val
item[1], item[2] = iota_row, iota_val
self.rows[item_row] = iota
self.rows[iota_row] = item
return 2
return 0
降级命令与升级命令几乎相同，因此为了简洁起见，我将省略它。请注意，这仅取决于 python 的列表、字典和 heapq 库。
现在使用 PQueue 数据类型，我们希望允许与实例进行分布式交互。gevent是一个很棒的库。尽管 gevent 相对较新并且仍处于测试阶段，但它的速度非常快并且经过了充分的测试。使用 gevent，我们可以很容易地设置一个监听 localhost:4040 的套接字服务器。这是我的服务器代码：
pqueue = PQueue()
def pqueue_server(sock, addr):
text = sock.recv(1024)
cmds = text.split(' ')
if cmds[0] == 'append':
result = pqueue.append(cmds[1])
elif cmds[0] == 'take':
result = pqueue.take()
elif cmds[0] == 'promote':
result = pqueue.promote(int(cmds[1]))
elif cmds[0] == 'demote':
result = pqueue.demote(int(cmds[1]))
else:
result = ''
sock.sendall(str(result))
print 'Request:', text, '; Response:', str(result)
if args.listen:
server = StreamServer(('127.0.0.1', 4040), pqueue_server)
print 'Starting pqueue server on port 4040...'
server.serve_forever()
在生产中运行之前，您当然需要进行一些更好的错误/缓冲区处理。但它对于快速原型制作来说效果很好。请注意，这不需要对 pqueue 对象进行任何锁定。Gevent 实际上并不并行运行代码，它只是给人一种印象。缺点是更多的核心无济于事，但好处是无锁代码。
不要误会我的意思，gevent SocketServer 将同时处理多个请求。但它通过协作多任务处理在回答请求之间进行切换。这意味着您必须放弃协程的时间片。虽然 gevents 套接字 I/O 函数被设计为屈服，但我们的 pqueue 实现却不是。幸运的是，pqueue 很快就完成了它的任务。
在制作原型时，我发现拥有客户也很有用。编写客户端需要一些谷歌搜索，所以我也将分享该代码：
if args.client:
while True:
msg = raw_input('> ')
sock = gsocket.socket(gsocket.AF_INET, gsocket.SOCK_STREAM)
sock.connect(('127.0.0.1', 4040))
sock.sendall(msg)
text = sock.recv(1024)
sock.close()
print text
要使用新的数据存储，请先启动服务器，然后启动客户端。在客户端提示下，您应该能够执行以下操作：
> append one
1
> append two
2
> append three
3
> promote 2
2
> promote 2
0
> take
two
考虑到您对数据存储的思考，您似乎真正关心的是吞吐量和持久性。但“扩展得非常好”并不能量化您的需求。所以我决定用测试函数对上述内容进行基准测试。这是测试功能：
def test():
import time
import urllib2
import subprocess
import random
random = random.Random(0)
from progressbar import ProgressBar, Percentage, Bar, ETA
widgets = [Percentage(), Bar(), ETA()]
def make_name():
alphabet = 'abcdefghijklmnopqrstuvwxyz'
return ''.join(random.choice(alphabet)
for rpt in xrange(random.randrange(3, 20)))
def make_request(cmds):
sock = gsocket.socket(gsocket.AF_INET, gsocket.SOCK_STREAM)
sock.connect(('127.0.0.1', 4040))
sock.sendall(cmds)
text = sock.recv(1024)
sock.close()
print 'Starting server and waiting 3 seconds.'
subprocess.call('start cmd.exe /c python.exe queue_thing_gevent.py -l',
shell=True)
time.sleep(3)
tests = []
def wrap_test(name, limit=10000):
def wrap(func):
def wrapped():
progress = ProgressBar(widgets=widgets)
for rpt in progress(xrange(limit)):
func()
secs = progress.seconds_elapsed
print '{0} {1} records in {2:.3f} s at {3:.3f} r/s'.format(
name, limit, secs, limit / secs)
tests.append(wrapped)
return wrapped
return wrap
def direct_append():
name = make_name()
pqueue.append(name)
count = 1000000
@wrap_test('Loaded', count)
def direct_append_test(): direct_append()
def append():
name = make_name()
make_request('append ' + name)
@wrap_test('Appended')
def append_test(): append()
...
print 'Running speed tests.'
for tst in tests: tst()
我对笔记本电脑上运行的服务器进行了 6 次测试。我认为结果非常好。这是输出：
Starting server and waiting 3 seconds.
Running speed tests.
100%|############################################################|Time: 0:00:21
Loaded 1000000 records in 21.770 s at 45934.773 r/s
100%|############################################################|Time: 0:00:06
Appended 10000 records in 6.825 s at 1465.201 r/s
100%|############################################################|Time: 0:00:06
Promoted 10000 records in 6.270 s at 1594.896 r/s
100%|############################################################|Time: 0:00:05
Demoted 10000 records in 5.686 s at 1758.706 r/s
100%|############################################################|Time: 0:00:05
Took 10000 records in 5.950 s at 1680.672 r/s
100%|############################################################|Time: 0:00:07
Mixed load processed 10000 records in 7.410 s at 1349.528 r/s
最后，耐用性是我没有完全原型化的唯一问题。但我也不认为这有多难。在我们的优先级队列中，项目堆（列表）包含将数据类型保存到磁盘所需的所有信息。由于使用 gevent，我们还可以以多处理方式生成函数，因此我想象使用这样的函数：
def save_heap(heap, toll):
name = 'heap-{0}.txt'.format(toll)
with open(name, 'w') as temp:
for val in heap:
temp.write(str(val))
gevent.sleep(0)
并将保存函数添加到我们的优先级队列中：
def save(self):
heap_copy = tuple(self.heap)
toll = self.toll
gevent.spawn(save_heap, heap_copy, toll)
现在，您可以每隔几分钟复制一次分叉并将数据存储写入磁盘的 Redis 模型。如果您需要更高的耐用性，请将上述内容与将命令记录到磁盘的系统结合起来。这些就是 Redis 使用的 AFP 和 RDB 持久性方法。