Help us shape the future of Go! Take the 2026 Go Developer Survey

tp

package module
v0.4.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 29, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Overview

Package tp - высокоуровневая библиотека для работы с очередями на базе queued: аналог перлового MR::TP.

Описание проекта, режимы работы и конфигурация - в README: https://github.com/adventures-team/go-tp/blob/master/docs/README.md

Index

Constants

View Source 
const DefaultProducerInstance = "producer"

DefaultProducerInstance - имя инстанса в онлайнконфе по умолчанию для продюсеров (веб-сервисы, и т.д.)

View Source 
const MaxVersion = queuedproto.VersionMask

MaxVersion - максимальная версия события (в верхних битах хранятся флаги)

View Source 
const SignalReload = syscall.SIGHUP

SignalReload - перечитывается конфиг (как и при изменении онлайнконфа)

Variables

View Source 
var (
	// ErrUnknownType - тип данных не зарегистрирован при помощи Register
	ErrUnknownType = errors.New("unregistered type")

	// ErrUnknownQueue - очередь не зарегистрирована при помощи [Register]
	ErrUnknownQueue = errors.New("unregistered queue")

	// ErrUnknownVersion - версия события не зарегистрирована при помощи [Register]
	ErrUnknownVersion = errors.New("unregistered version")

	// ErrVersionTooBig - версия не должна превышать MaxVersion
	ErrVersionTooBig = errors.New("version number exceeds MaxVersion")

	// ErrInternalHandler - тип структуры юзерского события должен поддерживать pHandler], т.к. [Register] принимает [Handler]. Говорит о серьёзной ошибке внутри tp
	ErrInternalHandler = errors.New("internal error: handlerInfo.dataType is not Handler")

	// ErrHardMemoryLimit - достигнут лимит по памяти oom_hard_limit
	ErrHardMemoryLimit = errors.New("memory usage is over the hard limit, stopping processing")

	// ErrPointerRequired - методу tp.Next требуется указатель на ресивер метода Handler (т.е. указатель на указатель на объект юзерского события)
	ErrPointerRequired = errors.New("tp.Next: pointer to Handle method's receiver is required")

	// ErrEventTypesDiffer - в цепочке событий типы различаются (скорее всего, внутренняя ошибка в eventListFromItems)
	ErrEventTypesDiffer = errors.New("tp.Next: event types differ")

	// ErrMisconstructedEvent - не надо инициализировать внутренности событий вручную.
	// Скорее всего, установили next = (*MyEvent)(nil) вместо Handler(nil)
	ErrMisconstructedEvent = errors.New("tp.Next: misconstructed event")

	// ErrStructRequired - для использования автозаполнения, тип события должен быть структурой/указателем на структуру
	ErrStructRequired = errors.New("struct type is required for autofill feature")

	// ErrExportedRequired - для использования автозаполнения требуется экспортируемое (публичное) поле
	ErrExportedRequired = errors.New("exported (public) field is required for autofill feature")

	// ErrFieldNotFound - не найдено поле с типом аргумента autofill
	ErrFieldNotFound = errors.New("no field found")
)
View Source 
var SignalsExit = [...]os.Signal{syscall.SIGTERM, syscall.SIGINT}

SignalsExit - graceful exit (дожидается завершения обработки всех событий, и выходит)

Functions

func CalcShard 

func CalcShard(eventID uint64, shardNum uint) uint

CalcShard - см. queuedclient.CalcShard.

func EncodeEvent 

func EncodeEvent(event Event) (eventData []byte, err error)

EncodeEvent кодирует событие.

Если нужно только отправлять события, регистрация не требуется (в отличие от предыдущих версий tp).

При соответствии интерфейсам Builder и/или EventIDBuilder вызываются соответствующие методы.

Кодировщику безразлично, передаётся указатель или значение, но для вызова методов по указанным интерфейсам нужно соблюдать правильный тип (тот, для которого объявлен метод - очевидно, для Builder нужен указатель).

ID события, кроме интерфейса EventIDBuilder, можно также передать явно. Номер шарда будет вычислен из ID события, если оно задано одним из этих способов. Если EventID.ID == 0, шард будет выбран по round robin (queuedclient.BalanceShard).

func Register 

func Register[EvType Event, EvSlice ~[]EvType](q Queue, handler func(context.Context, EvSlice) error) error

Register регистрирует обработчик очереди для получения событий. Для отправки (Publish) регистрация не требуется. handler принимает слайс событий, поддерживающих интерфейс Event (конкретных типов - не слайс интерфейсов). ID очереди и версия события определяется методами QueueID и Version. Для одной очереди могут быть зарегистрированы обработчики событий нескольких версий.

Types

type BaseEvent 

type BaseEvent struct {
	// contains filtered or unexported fields
}

BaseEvent - базовый тип события, должен встраиваться в пользовательские типы событий.

func NewBaseEvent 

func NewBaseEvent() BaseEvent

NewBaseEvent - для использования в тестах. Возвращает проинициализированный BaseEvent, в котором будут работать флаги (в частности, метод Done). Для отправки инициализировать событие не нужно, функция предусмотрена только для тестов, которым нужно проверить установку Done.

func (BaseEvent) AddonID 

func (BaseEvent) AddonID(incompatible)

AddonID - не поддерживается. Тип BaseEvent не предназначен для передачи по протоколу

func (BaseEvent) Defer 

func (bev BaseEvent) Defer(ctx context.Context, delay time.Duration) error

Defer устанавливает задержку события.

Время задержки устанавливается сразу же. По завершении цикла обработки событиям, для которых не был вызван метод Done либо Defer, проставляется задержка по умолчанию.

Если событие было помечено как выполненное при помощи Done, а потом был вызван Defer, пометка на удаление снимается.

func (BaseEvent) Done 

func (bev BaseEvent) Done()

Done помечает событие обработанным.

События удаляются не сразу, а по завершении цикла обработки всех событий, полученных за один запрос из queued. В случае возникновения паник (в т.ч. необработанных), удаление все равно должно произойти до выхода из tp.Queue.Work

func (BaseEvent) GetEventID 

func (bev BaseEvent) GetEventID() queuedproto.EventID

GetEventID - возвращает ID события. Код, работающий с конкретным событием (не через интерфейс) может обращаться к полу EventID напрямую.

func (BaseEvent) HasCustomDefer 

func (bev BaseEvent) HasCustomDefer() bool

HasCustomDefer - установлено ли явно время повторной активации вызовом методов BaseEvent.Defer или Queue.Replace

func (BaseEvent) IsDone 

func (bev BaseEvent) IsDone() bool

IsDone - вызывался ли метод Done для события.

func (BaseEvent) MarshalAddon 

func (BaseEvent) MarshalAddon(incompatible)

MarshalAddon - не поддерживается. Тип BaseEvent не предназначен для передачи по протоколу

func (BaseEvent) MarshalIProto 

func (BaseEvent) MarshalIProto(incompatible)

MarshalIProto - не поддерживается. Тип BaseEvent не предназначен для передачи по протоколу

func (BaseEvent) MarshalerDisabled 

func (BaseEvent) MarshalerDisabled(incompatible)

func (BaseEvent) Queue 

func (bev BaseEvent) Queue() Queue

Queue возвращает интерфейс API очередей.

func (BaseEvent) Shard 

func (bev BaseEvent) Shard(shardNum uint) uint

Shard распределяет события по шардам в зависимости от хеша от ID события. Совместимо с перлом (алгоритм хеширования - crc от строкового десятичного представления eventID, распределение по шардам - остаток от деления).

func (BaseEvent) UnmarshalAddon 

func (BaseEvent) UnmarshalAddon(incompatible)

UnmarshalAddon - не поддерживается. Тип BaseEvent не предназначен для передачи по протоколу

func (BaseEvent) UnmarshalIProto 

func (BaseEvent) UnmarshalIProto(incompatible)

UnmarshalIProto - не поддерживается. Тип BaseEvent не предназначен для передачи по протоколу

type Builder 

type Builder interface {
	Build() error
}

Builder - при поддержке этого интерфейса типом события, метод Build вызывается при добавлении события в очередь. Можно использовать для автозаполнения часто используемых полей (время создания, etc). Аналог в перле - builder/default (для отдельных полей)

type DisableMarshaler 

type DisableMarshaler struct{}

DisableMarshaler - встраивание этого типа в структуру отключает сериализацию средствами iproto

func (DisableMarshaler) MarshalIProto 

func (DisableMarshaler) MarshalIProto(buf []byte) ([]byte, error)

MarshalIProto - не делает ничего

func (DisableMarshaler) MarshalerDisabled 

func (DisableMarshaler) MarshalerDisabled()

func (DisableMarshaler) UnmarshalIProto 

func (DisableMarshaler) UnmarshalIProto(buf []byte) ([]byte, error)

UnmarshalIProto - не делает ничего

type Event 

type Event interface {
	// Marshaler используется для кодирования события при отправке. Должен быть
	// реализован/сгенерён для типа события.
	// Если не реализовать/сгенерить Unmarshaler, обрабатываться такие события
	// не будут (их не получится зарегистрировать вызовом [Register]),
	// доступна будет только отправка (Publish)
	iproto.Marshaler

	// Version нужно определить для типа события.
	Version() VersionType

	// QueueID нужно определить для типа события.
	QueueID() QueueIDType

	// Shard принимает количество шардов, и должен возвращать номер
	// шарда queued, в который будет отправлено событие. Обычно
	// для этого используется ID события.
	//
	// В BaseEvent определён метод Shard, совместимый с перловой реализацией.
	// Тип события может переопределить метод шардирования.
	Shard(shardNum uint) uint

	// GetEventID возвращает ID события.
	//
	// Определён в BaseEvent.
	GetEventID() queuedproto.EventID

	// Queue возвращает интерфейс [Queue] для работы с API очередей.
	//
	// Определён в BaseEvent.
	Queue() Queue

	// Defer откладывает следующую активацию события на указзанную задержку.
	// Выполняется сразу, а не в конце обработки пачки событий (как это
	// происходит при вызове Done, или при дефолтном дефере).
	//
	// Определён в BaseEvent.
	Defer(context.Context, time.Duration) error

	// HasCustomDefer возвращает признак, был ли вызван метод [Defer] явно.
	//
	// Определён в BaseEvent.
	HasCustomDefer() bool

	// Done помечает событие успешно обработанным. Такие события удаляются из очереди
	// в конце цикла обработки пачки событий, полученных из очереди (под пачкой имеются
	// ввиду все события из одной очереди, включая все их версии - т.е. потенциально
	// больше, чем размер слайса, переданного обработчику). Удаление выполняется
	// одним запросом.
	//
	// Определён в BaseEvent.
	Done()

	// IsDone возвращает признак успешного выполнения события.
	//
	// Определён в BaseEvent.
	IsDone() bool
}

Event - интерфейс обработчика очереди. Структура обработчика должна встраивать тип BaseEvent.

Нужно определить два метода - Version и QueueID, остальные унаследуются от встроенного BaseEvent.

type EventIDBuilder 

type EventIDBuilder interface {
	BuildEventID() uint64
}

EventIDBuilder - события, поддерживающие этот интерфейс, могут генерить ID события сами (аналог в перле - generate_id). По дефолту, ID событий генерит queued. Поля, автозаполненные при помощи Build(), можно использовать в методе BuildEventID()

type EventWithCreationTime 

type EventWithCreationTime interface {
	Event
	GetCreationTime() addons.CreationTime
}

EventWithCreationTime - событие, встраивающее addons.CreationTime.

Поддерживает middleware addons.CreationTime

type Handler 

type Handler[EvType Event, EvSlice ~[]EvType] func(context.Context, EvSlice) error

func WithMaxAge 

func WithMaxAge[EvType EventWithCreationTime, EvSlice ~[]EvType](handler Handler[EvType, EvSlice]) Handler[EvType, EvSlice]

WithMaxAge - ограничивает время обработки события значением max_age, которое может задаваться в конфиге для очереди, инстанса или глобально.

Пример: 

tp.Register(q, tp.WithMaxAge(func (ctx context.Context, events []MyEvent) error {
    // обычный обработчик
    for _, event := range events {
        // события старше max_age сюда не попадут
    }
    return nil
}))

func WithRetry 

func WithRetry[EvType RetryableEvent, EvSlice ~[]EvType](handler Handler[EvType, EvSlice], opts ...RetryOption) Handler[EvType, EvSlice]

WithRetry - ограничивает количество попыток обработки события.

При помощи опций можно задать коллбеки:

  • OnRetry - вызывается после увеличения счётчика событий в случае неудачной обработки (вызова метода Retry). После последней попытки не вызывается.
  • OnRetryExhausted - вызывается после последней неудачной попытки обработки.

Настройки в onlineconf (задаются для очереди или глобально): 

retry_limit - максимальное количество попыток, после которого событие удаляется. Формат - int
retry_type  - тип задержки - const или constant - постоянная, lin или linear - линейная, exp или exponential - экспоненциальная
retry_delay - постоянная часть задержки (для всех типов). Формат - [time.ParseDuration]
retry_coeff - коэффициент (для линейной или экспоненциальной задержки). Формат - [time.ParseDuration]
retry_exp   - экспонента. Формат - float. Значение по умолчанию - 2.

Задержка вычисляется по формуле:
  constant:   retry_delay
  linear:     retry_delay + retry_coeff * retryCount
  exponential: retry_delay + retry_coeff * retry_exp ^ retryCount

где Count - номер текущей попытки (начиная с 0)

Пример: 

type MyEvent struct {
    addons.Retry
    UserID uint64
    // ...
}

tp.Register(q, tp.WithRetry(func (ctx context.Context, events []MyEvent) error {
    for _, event := range events {
        data, err := SelectUserByID(event.UserID)
        if err != nil {
            continue // пытаемся обработать событие неограниченное кол-во раз
                     // с постоянной задержкой defer_time.
                     // после выхода из обработчика для всех необработанных событий
                     // будет вызван Defer (одним запросом)
        }

        resp, err := SendToRemoteSystem(ctx, data)
        if err != nil {
            event.Retry() // пытаемся обработать указанное в конфиге кол-во раз
                          // задержка и её тип указаны в конфиге (retry_*)
        }

        // ...
        event.Done() // успешная обработка, после выхода из обработчика
                     // все обработанные события удаляются (одним запросом)
    }
}))

func WithRetry2 

func WithRetry2[EvType RetryableEvent2, EvSlice ~[]EvType](handler Handler[EvType, EvSlice], opts ...Retry2Option) Handler[EvType, EvSlice]

type IgnoreDuplicateItems 

type IgnoreDuplicateItems interface {
	IgnoreDuplicateItems() bool
}

IgnoreDuplicateItems - для событий, вернувших true, игнорируются дубли по ID сообщения (выставляется флаг FlagIgnoreWrongItem). Следует использовать вместе с EventIDBuilder, если нужна идемпотентность ID. Перловый аналог называется так же.

type InitOption 

type InitOption func(*InitOptions)

InitOption - WithInstanceName, WithLogger, WithMetrics, WithLogLevelConfig

func WithInstanceName 

func WithInstanceName(instanceName string) InitOption

WithInstanceName устанавливает имя инстанса.

По умолчанию, для обработчиков читается из командной строки из флага -instance, для продюсеров - DefaultProducerInstance ("producer")

func WithLogLevelConfig 

func WithLogLevelConfig(lc advslog.LevelConfig) InitOption

WithLogLevelConfig включает пер-пакетные уровни логгирования из указанного поддерева онлайнконфа (advslog.WithLevelConfig) - например, module.Subtree("/tp/go/log"). Оверрайд для пакета задаётся ключом "<root>/<import-path>/level", в частности "/tp/go/log/github.com/adventures-team/go-iprotoconn/level" задаёт уровень логгирования iproto-транспорта независимо от прикладных уровней.

Правила использования:

  • НЕ создавайте ключ "<root>/level": глобальным уровнем владеет параметр log_level (применяется из applyConfig) - иначе за глобальный уровень начнут конкурировать два писателя;
  • поддерево должно раздаваться с child lists (ключи "<path>/" со списками детей) - без них advslog не найдёт оверрайды и не подпишется на изменения;
  • опция действует, только когда tp сам строит логгер по умолчанию: вместе с WithLogger она игнорируется (владелец логгера сам решает, как его собрать);
  • advslog.Init выполняется один раз на процесс: если приложение уже вызывало его без опций, поднастройка не применится - опция должна попасть в самый первый Init.

func WithLogger 

func WithLogger(logger *slog.Logger) InitOption

WithLogger устанавливает логгер.

По умолчанию, вызывается advslog.Init.

func WithMetrics 

func WithMetrics(ms advmetricsset.Set) InitOption

WithMetrics подключает сбор метрик.

type InitOptions 

type InitOptions struct {
	// contains filtered or unexported fields
}

InitOptions - тип напрямую использовать не надо. Не сделан приватным для упрощения написания тестов.

type MarshalerDisabled 

type MarshalerDisabled interface {
	MarshalerDisabled()
}

MarshalerDisabled - для проверки, отключен ли у типа маршалер средствами DisableMarshaler

type Middleware 

type Middleware[EvType Event, EvSlice ~[]EvType] func(Handler[EvType, EvSlice]) Handler[EvType, EvSlice]

type Queue 

type Queue interface {
	// Client возвращает интерфейс низкоуровневого API очередей
	Client() QueuedClient

	// Config возвращает высокоуровневый конфиг очередей с возможностью определения значений
	// для каждой очереди, с возможностью задания глобальных и per-instance дефолтов.
	Config() *queueconfig.Config

	// NewPoller возвращает интерфейс для опроса событий из очереди.
	// Вынесен для возможности внешней реализации разных стратегий опроса очереди
	// (например, managed master).
	NewPoller(logger *slog.Logger) workerpool.Poller

	// IsRegistered проверяет, зарегистрирована ли указанная очередь вызовом Register.
	IsRegistered(queueID uint16) bool

	// Publish кодирует событие при помощи метода Encode, и отправляет его в очередь методом [queuedclient.Publish].
	// Клиент очередей должен быть проинициализирован.
	//
	// Возвращается ID события (сгенерённый методом BuildEventID(), либо возвращённый queued).
	//
	// Если у события есть метод IgnoreDuplicateItems, и он возвращает true, дубли при
	// совпадении ID события игнорируются (также подавляется ошибка в логе queued флагом queuedproto.FlagIgnoreWrongItem).
	Publish(ctx context.Context, event Event, opts ...queuedclient.PublishOption) (queuedproto.EventID, error)

	// Replace заменяет существующее событие в очереди. ID события должно быть обязательно задано.
	// Если время срабатывания не установлено через опции, оно не изменяется.
	Replace(ctx context.Context, event Event, opts ...queuedclient.PublishOption) error

	MetricsSet() advmetricsset.Set
	SetMetricsSet(advmetricsset.Set)
}

Queue - высокоуровневое API очередей.

В комментариях описано поведение реализации, возвращаемой NewQueue.

func InitProducer 

func InitProducer[T queueconfig.OnlineConf](conf queueconfig.OnlineConfWithSubtree[T], opts ...InitOption) Queue

InitProducer - инициализация типового продюсера (веб-сервис, и т.д.): 

func main() {
	conf, err := onlineconf.OpenSubtree("TREE", "/project/queued")
	queue := tp.InitProducer(conf)

	err := queue.Register(QueueID, &QueueEvent{}, nil)
	if err != nil { zLog.Fatal().Err(err).Send() }

	err = queue.Register(AnotherQueueID, &AnotherQueueEvent{}, nil)
	if err != nil { zLog.Fatal().Err(err).Send() }

	svc := NewService(myServiceConfig{
		...
		Queue: queue,
		...
	})

	...

	svc.Serve()
}

func (svc *Service) PostUserDataHandler(...) {
	...
	svc.Queue.Publish(&QueueEvent{
		UserID: params.UserID(),
		...
	})

	...
}

Опции:

В случае ошибки инициализации происходит выход с логгированием на уровне FATAL. Предполагается вызов из функции main.

func NewQueue 

func NewQueue(ctx context.Context, conf *queueconfig.Config, opts ...queuedclient.DialOption) (Queue, error)

NewQueue создаёт объект для высокоуровневой работы с API очередей. Инициализирует клиент очередей (queuedclient) с указанным конфигом.

Опции передаются в queuedclient.Dial; в частности, queuedclient.WithMetricsSet включает транспортные метрики iproto_* (реестр подключается при создании клиента).

type QueueIDType 

type QueueIDType = uint16

QueueIDType - тип id очереди. Используем type alias - просто для удобства восприятия, "настоящий" тип (с методами, и громоздким приведением типов) нам не нужен.

type QueuedClient 

type QueuedClient interface {
	// Publish отправляет событие в очередь.
	// Автоинкремент ID и способы задания задержки описаны в документации к пакету queuedclient.
	Publish(ctx context.Context, queueID uint16, eventID queuedproto.EventID, flags uint32, eventData []byte, opts ...queuedclient.PublishOption) (queuedproto.EventID, error)

	// Replace заменяет существующее событие в очереди.
	Replace(ctx context.Context, queueID uint16, eventID queuedproto.EventID, eventData []byte, opts ...queuedclient.PublishOption) (bool, error)

	// Defer устанавливает задержку обработки для указанных событий. Если задержка не указана (delay == 0), используется значение defer_time из конфига.
	Defer(ctx context.Context, queueID uint16, delay time.Duration, eventIDs []queuedproto.EventID) error

	// Delete удаляет указанные события параллельно изо всех необходимых шардов пачками по update_batch_size событий.
	Delete(ctx context.Context, queueID uint16, eventIDs []queuedproto.EventID) error

	// GetActive получает активные события из очереди queueID изо всех шардов одновременно.
	GetActive(ctx context.Context, queueID uint16, count int) (queuedproto.ItemList, error)

	// GetByIDs запрашивает события из очереди queueID с указанными ID из указанных шардов параллельно.
	GetByIDs(ctx context.Context, queueID uint16, eventIDs []queuedproto.EventID) (queuedproto.ItemList, error)

	// GetByIDsBroadcast запрашивает события из очереди queueID с указанными ID изо всех шардов одновременно.
	GetByIDsBroadcast(ctx context.Context, queueID uint16, eventIDs []uint64) (queuedproto.ItemList, error)

	// SetActivationTime устанавливает время активации для указанных событий.
	SetActivationTime(ctx context.Context, queueID uint16, atime time.Time, eventIDs []queuedproto.EventID) error

	// QueueStat запрашивает статистику очереди queueID изо всех шардов одновременно (результат - по шардам).
	QueueStat(ctx context.Context, queueID uint16) ([]queuedproto.RespQueueStat, error)

	// ShardNum возвращает количество сконфигурированных шардов queued.
	ShardNum() uint

	// MakeBroadcastRequest посылает запрос req параллельно во все шарды.
	MakeBroadcastRequest(ctx context.Context, req queuedproto.Request, resFunc queuedclient.ResultFunc) error

	// MakeMulticastRequest посылает запросы для указанных eventIDs в нужные шарды.
	MakeMulticastRequest(ctx context.Context, batchSize int, eventIDs []queuedproto.EventID, reqFunc queuedclient.RequestFunc, resFunc queuedclient.ResultFunc) error

	// MakeUnicastRequest посылает запрос req к указанному шарду.
	MakeUnicastRequest(ctx context.Context, shard uint, req queuedproto.Request) (usedShard uint, respBytes []byte, err error)

	MetricsSet() advmetricsset.Set
	SetMetricsSet(advmetricsset.Set)

	// IsHealthy сообщает о работоспособности кластера queued: ни у одного шарда
	// не открыт circuit breaker, и в каждом шарде есть пригодный сервер.
	// Холодный старт (соединения ещё не установлены) считается здоровым состоянием.
	IsHealthy() bool

	// IsOnline проверяет доступность кластера queued.
	//
	// Deprecated: эквивалентен IsHealthy (живые TCP-соединения больше не проверяются,
	// go-iprotoconn устанавливает их по требованию и в фоне) - используйте IsHealthy.
	IsOnline() bool
}

QueuedClient - интерфейс низкоуровневого API очередей (клиента queued). Реализован в queuedclient.QueuedClient (там же - более подробное описание методов).

type Retry2Option 

type Retry2Option func(*retry2Opt)

func OnRetry2 

func OnRetry2[T RetryableEvent2](f func(T)) Retry2Option

OnRetry2 аналог OnRetry для WithRetry2

func OnRetry2Exhausted 

func OnRetry2Exhausted[T RetryableEvent2](f func(T, string)) Retry2Option

OnRetry2Exhausted аналог OnRetryExhausted для WithRetry2. В обработчик так же передаётся имя счётчика — cnt

func Retry2CounterNames 

func Retry2CounterNames(retry1, retry2 string) Retry2Option

Retry2CounterNames позволяет задать альтернативные имена для настроек счётчиков в онлайнконфе. По умолчанию, "1" и "2".

type RetryOption 

type RetryOption func(*retryOpt)

RetryOption - OnRetry, OnRetryExhausted

func OnRetry 

func OnRetry[T RetryableEvent](f func(T)) RetryOption

OnRetry определяет пользовательский обработчик, который будет вызван при ретрае. Используется для реализации кастомной обработки события при ретрае, будь то логгирование, отправка метрики и т.п.

func OnRetryExhausted 

func OnRetryExhausted[T RetryableEvent](f func(T)) RetryOption

OnRetryExhausted определяет пользовательский обработчик, который будет вызван при исчерпании лимита ретраев. Используется для реализации кастомной обработки события при ретрае, будь то логгирование, отправка метрики и т.п.

type RetryableEvent 

type RetryableEvent interface {
	Event
	IncRetryCount()
	GetRetryCount() uint32
	Retry()
	NeedRetry() bool
}

RetryableEvent - событие, встраивающее addons.Retryable. Поддерживает middleware WithRetry.

type RetryableEvent2 

type RetryableEvent2 interface {
	Event
	IncRetry1Count()
	IncRetry2Count()
	GetRetry1Count() uint32
	GetRetry2Count() uint32
	Retry1()
	Retry2()
	NeedRetry1() bool
	NeedRetry2() bool
}

RetryableEvent2 - событие, встраивающее addons.Retryable2. Поддерживает middleware WithRetry2.

type StandaloneMaster 

type StandaloneMaster struct {
	Queue
	// contains filtered or unexported fields
}

StandaloneMaster - управление пулом обработчиков очередей при помощи onlineconf. Поддерживает все методы типа Queue.

Не путать с перловым менеджером очередей.

func InitProcessor 

func InitProcessor[T queueconfig.OnlineConf](conf queueconfig.OnlineConfWithSubtree[T], opts ...InitOption) *StandaloneMaster

InitProcessor - инициализация типового обработчика очередей. Простейший случай: 

func main() {
	conf, err := onlineconf.OpenSubtree("TREE", "/project/queued")
	qm := tp.InitProcessor(conf)
	app := NewService(myServiceConfig{...})

	err := qm.Register(QueueID, &QueueEvent{}, app)
	if err != nil { zLog.Fatal().Err(err).Send() }

	err = qm.Register(AnotherQueueID, &AnotherQueueEvent{}, app)
	if err != nil { zLog.Fatal().Err(err).Send() }

	qm.Manage()
}

Опции:

  • WithInstanceName - по умолчанию - читается параметр командной строки -instance (flag.Parse() из main() в этом случае вызывать до вызова InitProcessor не следует).
  • WithLogger - если не указано явно, инициализируется при помощи advslog.Init.

В случае ошибки инициализации происходит выход с логгированием на уровне FATAL. Предполагается вызов из функции main.

func NewStandaloneManager 

func NewStandaloneManager(queue Queue, pool *workerpool.WorkerPool, logger *slog.Logger) *StandaloneMaster

NewStandaloneManager создаёт новый объект для управления пулом.

Это - низкоуровневый интерфейс. Для типового кейса следует юзать обёртку InitProcessor.

func (*StandaloneMaster) Run 

func (qm *StandaloneMaster) Run()

Run - для обработчиков очередей. Запуск управления пулом воркеров из конфига. Применяет конфиг сразу при вызове, а также при изменении конфига, и по сигналу SIGHUP. Если в конфиге при первом запуске обнаружены ошибки - пул воркеров останавливается, и функция возвращается. При изменении конфига/по сигналу выход/остановка происходят только при ошибке достижения oom_hard_limit. Для обычных ошибок - пишется ошибка в лог, и остаётся старый конфиг. Вызывать из main, при возврате - выходить.

При возврате (после остановки пула) горутина менеджера конфига тоже останавливается, Run дожидается её выхода (иначе она продолжала бы жить и логгировать после Stop - в тестах это приводило к панике "Log in goroutine after test has completed").

func (*StandaloneMaster) Stop 

func (qm *StandaloneMaster) Stop()

Stop - враппер для WorkerPool.Stop. Останавливает обработку новых событий, по завершению запущенных обработчиков, метод Manage выходит.

type Stat 

type Stat struct {
	// contains filtered or unexported fields
}

Stat is used by managed master. Not an application metrics.

func (*Stat) Errors 

func (s *Stat) Errors() uint32

func (*Stat) Processed 

func (s *Stat) Processed() uint32

type Unpacker 

type Unpacker interface {
	Unpack() error
}

Unpacker - если у события объявлен метод Unpack, он вызывается при поступлении события после его распаковки. Можно использовать для извлечения полей, запакованных в EventID.

type VersionType 

type VersionType = uint8 // из-за флагов в номере версии в перле, максимальный номер версии - 31 (queuedproto.FlagEnableAddons-1).

VersionType - id версии событий.

Source Files

View all Source files

Directories

Path Synopsis
cmd
queued-mover command
queued-mover переносит события очередей из одного кластера queued (источник, A) в другой (приёмник, B): читает активные события из шардов A, публикует их в B как есть (тело события не декодируется) и удаляет из A. Логика - пакет internal/mover, краткое описание - в docs/README.md.
 queued-mover переносит события очередей из одного кластера queued (источник, A) в другой (приёмник, B): читает активные события из шардов A, публикует их в B как есть (тело события не декодируется) и удаляет из A. Логика - пакет internal/mover, краткое описание - в docs/README.md.
queued-stresser command
queued-stresser - parallel and asynchronous queued stress test.
 queued-stresser - parallel and asynchronous queued stress test.
Package encodings - поддержка сериализации событий.
 Package encodings - поддержка сериализации событий.
mover
Package mover - логика утилиты queued-mover: перенос событий очередей из одного кластера queued (источник, A) в другой (приёмник, B) поверх queuedclient.
 Package mover - логика утилиты queued-mover: перенос событий очередей из одного кластера queued (источник, A) в другой (приёмник, B) поверх queuedclient.
managed
master
qmclient
stats
Package stats provides functionality for tracking and managing statistical data related to queues and slots.
 Package stats provides functionality for tracking and managing statistical data related to queues and slots.
Package queueconfig - доступ к конфигу с возможностью переопределения значений для очереди и инстанса.
 Package queueconfig - доступ к конфигу с возможностью переопределения значений для очереди и инстанса.
Package queuedclient - низкоуровневый клиент queued.
 Package queuedclient - низкоуровневый клиент queued.
examples command
mock/event
Package mockevent is a generated GoMock package.
 Package mockevent is a generated GoMock package.
mock/mover
Package mockmover is a generated GoMock package.
 Package mockmover is a generated GoMock package.
mock/onlineconf
Package mockonlineconf is a generated GoMock package.
 Package mockonlineconf is a generated GoMock package.
mock/poller
Package mockpoller is a generated GoMock package.
 Package mockpoller is a generated GoMock package.
mock/queue
Package mockqueue is a generated GoMock package.
 Package mockqueue is a generated GoMock package.
mock/slotpool
Package mockslotpool is a generated GoMock package.
 Package mockslotpool is a generated GoMock package.
mock/stats
Package mockstat is a generated GoMock package.
 Package mockstat is a generated GoMock package.
Package workerpool - абстрактный пул воркеров.
 Package workerpool - абстрактный пул воркеров.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.