Support delayed queues for redis

Implement blueprint delayed-queues

Change-Id: I6ee18df2c02ab11985ba4c35f8cbf815c4123a8b
This commit is contained in:
yangzhenyu 2017-10-18 14:05:51 +08:00
parent 39d5b2bc53
commit 8b071a46bb
4 changed files with 28 additions and 6 deletions

View File

@ -96,6 +96,8 @@ class MessageController(storage.Message, scripting.Mixin):
+---------------------+---------+
| created time | cr |
+---------------------+---------+
| delay expiry time | d |
+---------------------+---------+
4. Messages rank counter (Redis Hash):
@ -223,7 +225,7 @@ class MessageController(storage.Message, scripting.Mixin):
def _list(self, queue, project=None, marker=None,
limit=storage.DEFAULT_MESSAGES_PER_PAGE,
echo=False, client_uuid=None,
include_claimed=False,
include_claimed=False, include_delayed=False,
to_basic=True):
if not self._queue_ctrl.exists(queue, project):
@ -262,6 +264,10 @@ class MessageController(storage.Message, scripting.Mixin):
filters.append(functools.partial(utils.msg_claimed_filter,
now=now))
if not include_delayed:
filters.append(functools.partial(utils.msg_delayed_filter,
now=now))
if not echo:
filters.append(functools.partial(utils.msg_echo_filter,
client_uuid=client_uuid))
@ -344,10 +350,11 @@ class MessageController(storage.Message, scripting.Mixin):
def list(self, queue, project=None, marker=None,
limit=storage.DEFAULT_MESSAGES_PER_PAGE,
echo=False, client_uuid=None,
include_claimed=False):
include_claimed=False, include_delayed=False):
return self._list(queue, project, marker, limit, echo,
client_uuid, include_claimed)
client_uuid, include_claimed,
include_delayed)
@utils.raises_conn_error
@utils.retries_on_connection_error
@ -419,6 +426,7 @@ class MessageController(storage.Message, scripting.Mixin):
claim_id=None,
claim_expires=now,
claim_count=0,
delay_expires=now + msg.get('delay', 0),
body=msg.get('body', {}),
)

View File

@ -22,7 +22,8 @@ import msgpack
from oslo_utils import encodeutils
from oslo_utils import uuidutils
MSGENV_FIELD_KEYS = (b'id', b't', b'cr', b'e', b'u', b'c', b'c.e', b'c.c')
MSGENV_FIELD_KEYS = (b'id', b't', b'cr', b'e', b'u', b'c', b'c.e',
b'c.c', b'd')
SUBENV_FIELD_KEYS = (b'id', b's', b'u', b't', b'e', b'o', b'p', b'c')
@ -49,6 +50,7 @@ class MessageEnvelope(object):
'claim_id',
'claim_expires',
'claim_count',
'delay_expires',
]
def __init__(self, **kwargs):
@ -64,6 +66,7 @@ class MessageEnvelope(object):
_validate_uuid4(self.claim_id)
self.claim_expires = kwargs['claim_expires']
self.claim_count = kwargs.get('claim_count', 0)
self.delay_expires = kwargs.get('delay_expires', 0)
@staticmethod
def from_hmap(hmap):
@ -274,6 +277,7 @@ def _hmap_to_msgenv_kwargs(hmap):
'claim_id': claim_id,
'claim_expires': int(hmap[b'c.e']),
'claim_count': int(hmap[b'c.c']),
'delay_expires': int(hmap.get(b'd', 0)),
}
@ -287,6 +291,7 @@ def _msgenv_to_hmap(msg):
'c': msg.claim_id or '',
'c.e': msg.claim_expires,
'c.c': msg.claim_count,
'd': msg.delay_expires,
}

View File

@ -55,13 +55,16 @@ while (#claimed_msgs < limit) do
-- unclaimed as well.
if not found_unclaimed then
local msg = redis.call('HMGET', mid, 'c', 'c.e')
local msg = redis.call('HMGET', mid, 'c', 'c.e', 'd')
if msg[1] == false and msg[2] == false then
-- NOTE(Eva-i): It means the message expired and does not
-- actually exist anymore, we must later garbage collect it's
-- ID from the set and move on.
msg_ids_to_cleanup[#msg_ids_to_cleanup + 1] = mid
elseif msg[1] == '' or tonumber(msg[2]) <= now then
elseif (msg[1] == '' or tonumber(msg[2]) <= now)
and tonumber(msg[3]) <= now then
-- NOTE(cdyangzhenyu): If the message's delay time has not
-- expired, the message can not be claimed.
found_unclaimed = true
end
end

View File

@ -219,6 +219,12 @@ def msg_claimed_filter(message, now):
return message.claim_id and (now < message.claim_expires)
def msg_delayed_filter(message, now):
"""Return True IFF the message is currently delayed."""
return now < message.delay_expires
def msg_echo_filter(message, client_uuid):
"""Return True IFF the specified client posted the message."""