Architecture and Consumer Process

This document explains the architecture of the Trigger component's consumer process, which connects to MySQL binlog streams and processes database change events. It covers the Consumer class design, how ConsumerManager orchestrates multiple consumer processes, the integration with MySQLReplicationFactory, and the lifecycle of a consumer from startup to shutdown.

For information about event processing and trigger execution within the consumer, see Event Processing and Trigger Execution. For health monitoring and fault tolerance mechanisms, see Health Monitoring and Fault Tolerance.

---

Consumer Class Architecture

The Consumer class src/trigger/src/Consumer.php29-240 is the core component that manages a single MySQL binlog replication connection. Each consumer instance connects to one MySQL server, subscribes to binlog events, and coordinates the event processing pipeline.

Consumer Components and Dependencies

Sources: src/trigger/src/Consumer.php29-70 src/trigger/src/Config.php16-31

The Consumer constructor src/trigger/src/Consumer.php45-70 initializes:

Property	Type	Purpose	Default
connection	string	Database connection name	'default'
name	string	Process name for identification	trigger.{connection}
identifier	string	Coordinator identifier	trigger.{connection}
config	Config	Configuration wrapper	Constructed from options
binLogCurrentSnapshot	BinLogCurrentSnapshotInterface	Tracks binlog position	RedisBinLogCurrentSnapshot
healthMonitor	?HealthMonitor	Monitors replication health	Created if enabled
serverMutex	?ServerMutexInterface	Distributed lock for HA	Created if enabled

Sources: src/trigger/src/Consumer.php45-70 src/trigger/publish/trigger.php14-56

---

MySQLReplicationFactory Integration

The Consumer creates and configures a MySQLReplicationFactory instance from the php-mysql-replication library, which performs the actual MySQL binlog protocol communication.

Replication Factory Configuration

Sources: src/trigger/src/Consumer.php182-239

The makeReplication() method src/trigger/src/Consumer.php182-239 performs these steps:

1. Aggregate database and table filters src/trigger/src/Consumer.php186-194:

   • Merges databases_only from config with databases from TriggerManager
   • Merges tables_only from config with tables from TriggerManager

2. Build configuration src/trigger/src/Consumer.php196-208:

   • Connection parameters: user, host, password, port from config
   • Random slaveId between 10000-9999999
   • heartbeatPeriod for connection keepalive (default 3 seconds)
   • UUID for php-mysql-replication >= 8.0

3. Restore binlog position src/trigger/src/Consumer.php210-218:

   • Retrieves BinLogCurrent from snapshot
   • Sets binLogFileName and binLogPosition if snapshot exists
   • Logs position for debugging

4. Register subscribers src/trigger/src/Consumer.php228-236:

   • Gets subscriber classes for this connection from SubscriberManager
   • Instantiates each subscriber with consumer and logger dependencies
   • Registers them with the factory

Sources: src/trigger/src/Consumer.php182-239

---

Consumer Lifecycle

The consumer follows a structured lifecycle with distinct startup, running, and shutdown phases.

Startup and Main Loop

Sources: src/trigger/src/Consumer.php72-117

The start() method src/trigger/src/Consumer.php72-117 orchestrates the consumer lifecycle:

1. Reset state src/trigger/src/Consumer.php77:

   • Sets stopped = false

2. Start health monitor src/trigger/src/Consumer.php80:

   • Calls healthMonitor->process() if enabled
   • See Health Monitoring and Fault Tolerance

3. Create replication src/trigger/src/Consumer.php82:

   • Calls makeReplication() to create MySQLReplicationFactory

4. Resume coordinator src/trigger/src/Consumer.php85:

   • Signals coordinator with identifier to resume waiting coroutines

5. Register exit handler src/trigger/src/Consumer.php90-94:

   • Creates coroutine that waits for Constants::WORKER_EXIT
   • Calls stop() when worker exit signal received

6. Main consume loop src/trigger/src/Consumer.php96-109:

   • Checks isStopped() each iteration
   • Calls replication->consume() which blocks until binlog event arrives
   • Catches exceptions, logs error, and calls stop()

Sources: src/trigger/src/Consumer.php72-117

Stop and Cleanup

The stop() method src/trigger/src/Consumer.php171-175 performs cleanup:

• Sets stopped flag to true src/trigger/src/Consumer.php173
• Releases distributed lock if present src/trigger/src/Consumer.php174

Sources: src/trigger/src/Consumer.php171-175

---

ConsumerManager and Process Orchestration

The ConsumerManager src/trigger/src/ConsumerManager.php22-76 is responsible for creating and registering consumer processes based on configuration.

Process-Per-Connection Model

Sources: src/trigger/src/ConsumerManager.php22-76 src/trigger/src/Listener/BindTriggerProcessesListener.php22-47

Registration Process

The register() method src/trigger/src/ConsumerManager.php30-44 creates processes:

1. Read configuration src/trigger/src/ConsumerManager.php33:

   • Gets trigger.connections array from config
   • Each key is a connection name, value is connection options

2. Filter enabled connections src/trigger/src/ConsumerManager.php36-38:

   • Skips connections where enable is false

3. Create process src/trigger/src/ConsumerManager.php40:

   • Calls createProcess($connection, $options)

4. Register with ProcessManager src/trigger/src/ConsumerManager.php42:

   • Adds process to Hyperf's process registry

Sources: src/trigger/src/ConsumerManager.php30-44

Anonymous Process Creation

The createProcess() method src/trigger/src/ConsumerManager.php54-75 creates an anonymous process class:

Sources: src/trigger/src/ConsumerManager.php54-75

The anonymous process src/trigger/src/ConsumerManager.php56-74:

• Extends AbstractProcess src/trigger/src/ConsumerManager.php56
• Sets name from options or defaults to trigger{connection} src/trigger/src/ConsumerManager.php60
• Sets nums = 1 to ensure exactly one process per connection src/trigger/src/ConsumerManager.php61
• In handle(), creates a Consumer instance and calls start() src/trigger/src/ConsumerManager.php64-72

Sources: src/trigger/src/ConsumerManager.php54-75

---

Configuration Structure

The trigger system is configured through the trigger.php configuration file, organized by connection.

Configuration Schema

Section	Keys	Description	Default
connections.{name}	enable	Enable this connection	true
host	MySQL host	''
port	MySQL port	3306
user	MySQL user	''
password	MySQL password	''
databases_only	Database whitelist	[]
tables_only	Table whitelist	[]
heartbeat_period	Keepalive seconds	3
connect_retries	Retry attempts	10
subscribers	Subscriber classes	[TriggerSubscriber, SnapshotSubscriber]
server_mutex	enable	Enable distributed lock	true
prefix	Redis key prefix	{APP_NAME}:server_mutex:
expires	Lock TTL seconds	30
keepalive_interval	Renewal interval seconds	10
retry_interval	Acquisition retry seconds	10
health_monitor	enable	Enable health monitoring	true
interval	Monitor interval seconds	30
snapshot	version	Snapshot key version	1.0
expires	Snapshot TTL seconds	86400
interval	Snapshot interval seconds	10
concurrent	limit	Concurrent trigger limit	1

Sources: src/trigger/publish/trigger.php13-56

Dependency Injection Bindings

The ConfigProvider src/trigger/src/ConfigProvider.php14-48 registers these bindings:

Interface	Implementation	Purpose
ServerMutexInterface	RedisServerMutex	Distributed locking via Redis
BinLogCurrentSnapshotInterface	RedisBinLogCurrentSnapshot	Position persistence via Redis
Contract\LoggerInterface	StdoutLoggerInterface	Logging adapter

Sources: src/trigger/src/ConfigProvider.php24-28

---

Integration with Subscriber System

The consumer delegates event processing to registered subscribers. Each connection can have multiple subscribers that implement EventSubscriberInterface.

Subscriber Registration Flow

Sources: src/trigger/src/ConsumerManager.php30-44 src/trigger/src/Consumer.php228-236 src/trigger/src/SubscriberManager.php37-85 src/trigger/src/Listener/BindTriggerProcessesListener.php41-46

The default subscribers registered src/trigger/publish/trigger.php27-30 are:

1. TriggerSubscriber src/trigger/src/Subscriber/TriggerSubscriber.php31-221:

   • Subscribes to write, update, delete events
   • Dispatches to user-defined trigger classes
   • See Event Processing and Trigger Execution

2. SnapshotSubscriber src/trigger/src/Subscriber/SnapshotSubscriber.php17-42:

   • Subscribes to all events
   • Updates HealthMonitor with current BinLogCurrent
   • Enables position tracking for snapshots

Sources: src/trigger/src/Subscriber/TriggerSubscriber.php53-60 src/trigger/src/Subscriber/SnapshotSubscriber.php23-41

---

Consumer-to-Code Entity Mapping

This diagram shows how consumer concepts map to concrete code entities:

Sources: All files referenced in diagram labels

---

Process Isolation and Resource Management

Each consumer runs in an isolated process with dedicated resources:

Per-Consumer Resources

Resource	Scope	Implementation
MySQL connection	Per consumer	Created by MySQLReplicationFactory
Redis lock	Per connection name	RedisServerMutex with unique key
Binlog snapshot	Per connection	Redis key trigger:snapshot:binLogCurrent:{version}:{connection}
Health monitor timer	Per consumer	Timer instance in HealthMonitor
Event channel	Per consumer	Channel in TriggerSubscriber with configurable size
Concurrent limiter	Per consumer	Concurrent instance with configured limit

Sources: src/trigger/src/Consumer.php45-70 src/trigger/src/Mutex/RedisServerMutex.php36-59 src/trigger/src/Snapshot/RedisBinLogCurrentSnapshot.php65-74 src/trigger/src/Subscriber/TriggerSubscriber.php39-50

This isolation ensures:

• Independent failure domains - one connection failure doesn't affect others
• Resource isolation - each consumer has dedicated memory and connections
• Horizontal scalability - add connections without interference
• Independent restart - stop/start individual consumers without affecting others

Sources: src/trigger/src/ConsumerManager.php54-75