consumer

[CN] 开发者设计文档

This section is for this project maintainer only.

这个模块实现了 Consumer 的基类 BaseConsumer. 它实现的功能主要是从 stream 中 读 records, 如何处理 batch records, 是串行还是并行, 如何处理在消费 record 中的异常. 你的 stream system 可以是任何系统.

BaseConsumer 的子类是一些跟某些特定的 stream system 对接的 Consumer 的具体实现. 它们假设你可以用任何逻辑来消费 record. 它只是负责实现了 BaseConsumer.get_records() 方法.

class unistream.consumer.BaseConsumer(record_class: type[AbcRecord] = REQ, limit: int = REQ, checkpoint: BaseCheckPoint = REQ, exp_backoff_multiplier: int = REQ, exp_backoff_base: int = REQ, exp_backoff_min: int = REQ, exp_backoff_max: int = REQ, skip_error: bool = REQ, delay: int | float = REQ)

Consumer continuously fetches batch records from the stream system and process them.

Parameters:

• record_class – the record class.

• limit – the max number of records to fetch from the stream system.

• checkpoint – the BaseCheckpoint object for status tracking and fault tolerance.

• exp_backoff_multiplier – the multiplier of the exponential backoff

• exp_backoff_base – the base of the exponential backoff

• exp_backoff_min – the minimum wait time of the exponential backoff

• exp_backoff_max – the maximum wait time of the exponential backoff

• skip_error – if True, skip the error and continue to process the next record. this is the most common use case. if False, raise the error and stop the consumer.

• delay – the delay time between pulling two batches.

get_records(limit: int = None) → tuple[list[AbcRecord], str | int]

Get records from the stream system and determine the value of the next pointer for the next batch if we successfully process this batch of records.

Parameters:

limit – The maximum number of records to return.

Returns:

a two-item tuple. the first one is a list of records and the second one is the value of the next pointer for the next batch if we successfully process this batch of records.

Important

User has to implement this method.

If you need additional parameters other than the BaseConsumer built-in attributes and limit, you should extend this class and add the parameters to the subclass.

process_record(record: AbcRecord)

This method defines how to process a failed record.

Important

User has to implement this method.

process_failed_record(record: AbcRecord)

This method defines how to process a failed record.

Note

By default, it does nothing. In production, you should send this record to a dead-letter-queue (DLQ) for further investigation.

To avoid sending records to DLQ one by one, you can use get_not_succeeded_records() to get all failed records and send them to DLQ in batch.

Users can customize this method.

commit()

Mark the current batch has been fully processed.

process_batch(verbose: bool = False)

Note

Currently, we only support sequential processing.

run(verbose: bool = False)

Run the consumer.