Redis

Redis Streams

For more info on Redis Streams please visit https://redis.io/docs/latest/develop/data-types/streams/

RedisBroker

RedisBroker(
    *,
    redis: RedisClient,
    broker_config: RedisBrokerConfig | None = None,
    consumer_name: str,
    max_concurrency: int = 20
)

Parameters:

• redis (RedisClient) –

  Instance of redis

• broker_config (RedisBrokerConfig | None, default: None ) –

  Redis specific configuration

• consumer_name (str) –

  Name of stream consumer, if you run multiple workers you'd need to change that. https://redis.io/docs/latest/develop/data-types/streams/#consumer-groups and https://redis.io/docs/latest/develop/data-types/streams/#differences-with-kafka-tm-partitions

• max_concurrency (int, default: 20 ) –

  Max amount of tasks being concurrently added into redis stream

RedisBrokerConfig dataclass

RedisBrokerConfig(
    *,
    stream_name: str = "async-queue",
    maintenance_lock_name: str = "aiotaskqueue-maintenance-lock",
    maintenance_lock_timeout: timedelta = timedelta(
        minutes=10
    ),
    group_name: str = "default",
    xread_block_time: timedelta = timedelta(seconds=1),
    xread_count: int = 1,
    xtrim_interval: timedelta = timedelta(minutes=30)
)

stream_name class-attribute instance-attribute

stream_name: str = 'async-queue'

Stream name in redis (key name)

group_name class-attribute instance-attribute

group_name: str = 'default'

Redis stream group name, there usually shouldn't be a need to change it. See https://redis.io/docs/latest/commands/xgroup-create/

xread_block_time class-attribute instance-attribute

xread_block_time: timedelta = timedelta(seconds=1)

BLOCK parameter passed to redis XREAD command

xread_count class-attribute instance-attribute

xread_count: int = 1

Amount of entries to read from stream at once

xtrim_interval class-attribute instance-attribute

xtrim_interval: timedelta = timedelta(minutes=30)

Interval between XTRIM calls