目录
Celery是一个简单、灵活且可靠的,处理大量消息的分布式系统,专注于实时处理的异步任务队列,同时也支持任务调度。本文介绍 Celery 的负载均衡机制。
Autoscaler 的作用 实际就是在线调节进程池大小。这也和缓解负载相关,所以放在这里一起论述。
Celery 的负载均衡其实可以分为三个层次,而且是与 Kombu 高度耦合(本文 broker 以 Redis 为例)。
注意,这个顺序是从 worker 读取任务处理任务的角度 出发,而不是从系统架构角度出发。
因为从系统架构角度说,应该是 which worker ----> which queue in the worker ----> which subprocess in the worker
这个角度。
我们下面按照 "worker 读取任务处理任务角度" 的顺序进行分析。
Kombu 事实上是使用 redis 的 BRPOP 功能来完成对具体 queue 中消息的读取。
因为 task 可能会 用到多个 queue,所以具体从哪几个queue 读取?这时候就用到了策略。
Kombu 在每次监听时候,调用 _brpop_start 完成监听。其作用就是 选择下一次读取的queues。
_brpop_start 如下:
def _brpop_start(self, timeout=1):
# 得到一些内部queues
queues = self._queue_cycle.consume(len(self.active_queues))
if not queues:
return
# 得到queue对应的keys
keys = [self._q_for_pri(queue, pri) for pri in self.priority_steps
for queue in queues] + [timeout or 0]
self._in_poll = self.client.connection
self.client.connection.send_command('BRPOP', *keys) # 利用这些keys,从redis内部获取key的消息
此时变量如下:
self.active_queues = {set: 1} {'celery'}
len(self.active_queues) = {int} 1
self._queue_cycle = {round_robin_cycle} <kombu.utils.scheduling.round_robin_cycle object at 0x0000015A7EE9DE88>
self = {Channel} <kombu.transport.redis.Channel object at 0x0000015A7EE31048>
所以_brpop_start
就是从 self._queue_cycle 获得几个需要读取的queue。
具体如下图:
+
Kombu | Redis
|
|
+--------------------------------------------+ |
| Worker | |
| | | queue 1 key
| +-----------+ | |
| | queue 1 | | BRPOP(keys) |
| | queue 2 | keys | |
| | ...... | +--------+----------------------------------------> queue 2 key
| | queue n | ^ | |
| +-----------+ | keys | |
| | | |
| | | | queue 3 key
| +-------------+------------+ | |
| | Keys list | | |
| | | | |
| +--------------------------+ | |
+--------------------------------------------+ |
|
|
|
|
|
+
从上面代码中,我们可以知道 consume 就是返回 round_robin_cycle 中前几个 queue,即 return self.items[:n]。
而 self.items 的维护,是通过 rotate 完成的,就是把 最近用的 那个 queue 放到队列最后,这样给其他 queue 机会,就是 round robin 的概念了。
class round_robin_cycle:
"""Iterator that cycles between items in round-robin."""
def __init__(self, it=None):
self.items = it if it is not None else []
def update(self, it):
"""Update items from iterable."""
self.items[:] = it
def consume(self, n):
"""Consume n items."""
return self.items[:n]
def rotate(self, last_used):
"""Move most recently used item to end of list."""
items = self.items
try:
items.append(items.pop(items.index(last_used)))
except ValueError:
pass
return last_used
比如在如下代码中,当读取到消息之后,就会调用 self._queue_cycle.rotate(dest)
进行调整。
def _brpop_read(self, **options):
try:
try:
dest__item = self.client.parse_response(self.client.connection,
'BRPOP',
**options)
except self.connection_errors:
# if there's a ConnectionError, disconnect so the next
# iteration will reconnect automatically.
self.client.connection.disconnect()
raise
if dest__item:
dest, item = dest__item
dest = bytes_to_str(dest).rsplit(self.sep, 1)[0]
self._queue_cycle.rotate(dest) # 这里进行调整
self.connection._deliver(loads(bytes_to_str(item)), dest)
return True
else:
raise Empty()
finally:
self._in_poll = None
具体如下图:
+
Kombu | Redis
|
|
+--------------------------------------------+ |
| Worker | |
| | | queue 1 key
| +-----------+ | |
| | queue 1 | | BRPOP(keys) |
| | queue 2 | keys | |
| | ...... | +--------+----------------------------------------> queue 2 key
| | queue n | ^ | |
| +-----------+ | keys | |
| | | |
| + | | queue 3 key
| round_robin_cycle | |
| + | |
| | | |
| | | |
| +-------------+------------+ | |
| | Keys list | | |
| +--------------------------+ | |
+--------------------------------------------+ |
|
+
如果多个 worker 同时去使用 brpop 获取 broker 消息,那么具体哪一个能够读取到消息,其实这就是有一个 竞争机制,因为redis 的单进程处理,所以只能有一个 worker 才能读到。
这本身就是一个负载均衡。这个和 spring quartz 的负载均衡实现非常类似。
具体如下图:
+
Kombu | Redis
|
|
+--------------------------------------+ |
| Worker 1 | |
| | |
| +-----------+ | |
| | queue 1 | | BRPOP(keys) |
| | queue 2 | keys | |
| | ...... | +--------+-----------------------------+ |
| | queue n | ^ | | |
| +-----------+ | keys | | |
| | | | |
| + | | |
| round_robin_cycle | | | +--> queue 1 key
| ^ | | | |
| | | | | |
| | | | | Single Thread |
| +------------+---------+ | +---------------------> queue 2 key
| | keys list | | | | |
| +----------------------+ | | | |
+--------------------------------------+ | | |
| | +--> queue 3 key
| |
+--------------------------------------+ | |
| Worker 2 | BRPOP(keys) | |
| | +---------------+ |
| | | |
+--------------------------------------+ | |
| |
+--------------------------------------+ BRPOP(keys) | |
| Worker 3 | | |
| | +--------------+ |
| | +
| |
+--------------------------------------+
进程池中,使用了策略来决定具体使用哪一个进程来处理任务。
先讲解 strategy。在 AsynPool 启动有如下,配置了策略:
class AsynPool(_pool.Pool):
"""AsyncIO Pool (no threads)."""
def __init__(self, processes=None, synack=False,
sched_strategy=None, proc_alive_timeout=None,
*args, **kwargs):
self.sched_strategy = SCHED_STRATEGIES.get(sched_strategy,
sched_strategy)
于是我们看看 strategy 定义如下,基本由名字可以知道其策略意义:
SCHED_STRATEGY_FCFS = 1 # 先来先服务
SCHED_STRATEGY_FAIR = 4 # 公平
SCHED_STRATEGIES = {
None: SCHED_STRATEGY_FAIR,
'default': SCHED_STRATEGY_FAIR,
'fast': SCHED_STRATEGY_FCFS,
'fcfs': SCHED_STRATEGY_FCFS,
'fair': SCHED_STRATEGY_FAIR,
}
我们讲讲公平调度的概念。
不同系统对于公平调度的理解大同小异,我们举几个例子看看。
在 asynpool之中,有设置,看看"是否为 fair 调度":
is_fair_strategy = self.sched_strategy == SCHED_STRATEGY_FAIR
基于 is_fair_strategy 这个变量,Celery 的公平调度有几处体现。
在开始 poll 时候,如果是 fair,则需要 存在 idle worker 才调度,这样就给了 idler worker 一个调度机会。
def on_poll_start():
# Determine which io descriptors are not busy
inactive = diff(active_writes)
# Determine hub_add vs hub_remove strategy conditional
if is_fair_strategy:
# outbound buffer present and idle workers exist
add_cond = outbound and len(busy_workers) < len(all_inqueues)
else: # default is add when data exists in outbound buffer
add_cond = outbound
if add_cond: # calling hub_add vs hub_remove
iterate_file_descriptors_safely(
inactive, all_inqueues, hub_add,
None, WRITE | ERR, consolidate=True)
else:
iterate_file_descriptors_safely(
inactive, all_inqueues, hub_remove)
在具体发布 写操作 时候,也会看看是否 worker 已经正在忙于执行某一个 task,如果正在执行,就不调度,这样就给了其他 不忙worker 一个调度的机会。
def schedule_writes(ready_fds, total_write_count=None):
if not total_write_count:
total_write_count = [0]
# Schedule write operation to ready file descriptor.
# The file descriptor is writable, but that does not
# mean the process is currently reading from the socket.
# The socket is buffered so writable simply means that
# the buffer can accept at least 1 byte of data.
# This means we have to cycle between the ready fds.
# the first version used shuffle, but this version
# using `total_writes % ready_fds` is about 30% faster
# with many processes, and also leans more towards fairness
# in write stats when used with many processes
# [XXX On macOS, this may vary depending
# on event loop implementation (i.e, select/poll vs epoll), so
# have to test further]
num_ready = len(ready_fds)
for _ in range(num_ready):
ready_fd = ready_fds[total_write_count[0] % num_ready]
total_write_count[0] += 1
if ready_fd in active_writes:
# already writing to this fd
continue
if is_fair_strategy and ready_fd in busy_workers: # 是否调度
# worker is already busy with another task
continue
if ready_fd not in all_inqueues:
hub_remove(ready_fd)
continue
具体逻辑如下:
+
Kombu | Redis
|
BRPOP(keys) |
+------------------------------------+ |
| Worker 1 | +---------------+ |
| | | |
+------------------------------------+ | | queue 1 key
| | +->
| | |
+------------------------------------+ BRPOP(keys) | | Single thread |
| Worker 2 | +--------------------------------------> queue 2 key
| | | | (which worker) |
+------------------------------------+ | | |
| | |
+------------------------------------+ | | +-> queue 3 key
| Worker 3 | | |
| | | |
| +-----------+ | | |
| | queue 1 | | BRPOP(keys) | |
| | queue 2 | keys | | |
| | ...... | +--------+-------------------------+ |
| | queue n | ^ | |
| +-----------+ | keys | |
| | | |
| + | |
| round_robin_cycle (which queues) |
| ^ | |
| | | |
| | | |
| +----+----+ | |
| + |keys list| | |
| | +---------+ | |
+------------------------------------+ |
| |
| fair_strategy(which subprocess) |
| |
+-------+----------+----------------+ |
| | | |
v v v |
+-----+--------+ +------+-------+ +-----+--------+ |
| subprocess 1 | | subprocess 2 | | subprocess 3 | +
+--------------+ +--------------+ +--------------+
Autoscaler 的作用 实际就是在线调节进程池大小。这也和缓解负载相关,所以放在这里一起论述。
在 WorkerComponent 中可以看到,为 AutoScaler 注册了两个调用途径:
这样就会最大程度的加大调用频率。
class WorkerComponent(bootsteps.StartStopStep):
"""Bootstep that starts the autoscaler thread/timer in the worker."""
def create(self, w):
scaler = w.autoscaler = self.instantiate(
w.autoscaler_cls,
w.pool, w.max_concurrency, w.min_concurrency,
worker=w, mutex=DummyLock() if w.use_eventloop else None,
)
return scaler if not w.use_eventloop else None
def register_with_event_loop(self, w, hub):
w.consumer.on_task_message.add(w.autoscaler.maybe_scale) # 消费时候如果有需要,就会调整
hub.call_repeatedly( # 周期看看是否需要调整
w.autoscaler.keepalive, w.autoscaler.maybe_scale,
)
Autoscaler 是Background thread,这样 AutoScaler就可以在后台运行:
class bgThread(threading.Thread):
"""Background service thread."""
def run(self):
body = self.body
shutdown_set = self._is_shutdown.is_set
try:
while not shutdown_set():
body()
finally:
self._set_stopped()
Autoscaler 的定义如下,可以看到其逻辑就是定期判断是否需要调整:
class Autoscaler(bgThread):
"""Background thread to autoscale pool workers."""
def __init__(self, pool, max_concurrency,
min_concurrency=0, worker=None,
keepalive=AUTOSCALE_KEEPALIVE, mutex=None):
super().__init__()
self.pool = pool
self.mutex = mutex or threading.Lock()
self.max_concurrency = max_concurrency
self.min_concurrency = min_concurrency
self.keepalive = keepalive
self._last_scale_up = None
self.worker = worker
def body(self):
with self.mutex:
self.maybe_scale()
sleep(1.0)
def _maybe_scale(self, req=None):
procs = self.processes
cur = min(self.qty, self.max_concurrency)
if cur > procs:
self.scale_up(cur - procs)
return True
cur = max(self.qty, self.min_concurrency)
if cur < procs:
self.scale_down(procs - cur)
return True
def maybe_scale(self, req=None):
if self._maybe_scale(req):
self.pool.maintain_pool()
def update(self, max=None, min=None):
with self.mutex:
if max is not None:
if max < self.processes:
self._shrink(self.processes - max)
self._update_consumer_prefetch_count(max)
self.max_concurrency = max
if min is not None:
if min > self.processes:
self._grow(min - self.processes)
self.min_concurrency = min
return self.max_concurrency, self.min_concurrency
def scale_up(self, n):
self._last_scale_up = monotonic()
return self._grow(n)
def scale_down(self, n):
if self._last_scale_up and (
monotonic() - self._last_scale_up > self.keepalive):
return self._shrink(n)
def _grow(self, n):
self.pool.grow(n)
def _shrink(self, n):
self.pool.shrink(n)
def _update_consumer_prefetch_count(self, new_max):
diff = new_max - self.max_concurrency
if diff:
self.worker.consumer._update_prefetch_count(
diff
)
@property
def qty(self):
return len(state.reserved_requests)
@property
def processes(self):
return self.pool.num_processes