Problem Analysis

Consider a user registration promotional scenario where we introduce a message queue to decouple the main registration flow from promotional activities. The sequence diagram shows the account center completing user registration logic—writing to the account center database and sending a message to the MQ server before returning "registration successful" to the user. Two consumers then process the message asynchronously to deliver coupons or points.

This architectural approach is sound, but a critical question emerges: how do we ensure consistency between database writes and message sends? These two operations must either both succeed or both fail completely. We cannot tolerate a scenario where user data is successfully written to the database but message sending fails, as users would not receive coupons, leading to complaints and disputes.

This is fundamentally a distributed transaction problem—ensuring consistency between database writes and message sends across distributed operations.

Local Message Table + Scheduled Task Pattern

A common solution involves the "local message table + scheduled task" pattern.

First, create a local message table in the database with a structure similar to the following:

CREATE TABLE local_msg_record (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    account_no VARCHAR(64) NOT NULL,
    msg_status TINYINT DEFAULT 0 COMMENT '0=pending, 1=sent, 2=failed',
    retry_count INT DEFAULT 0,
    next_select_time DATETIME,
    create_time DATETIME,
    update_time DATETIME,
    INDEX idx_account_no (account_no),
    INDEX idx_create_time (create_time),
    INDEX idx_next_select_time (next_select_time)
);

After creating the message table, the user registration workflow becomes:

1. Begin database local transaction
2. Insert into user table
3. Insert into local_msg_record with account unique identifier and status (initial value 0)
4. Commit local transaction

This ensures transactional consistency between the user table and local_msg_record. If user information is successfully stored in the user table, a corresponding record must exist in local_msg_record. We only need to send corresponding MQ messages based on records in local_msg_record.

To prevent local_msg_record writes from causing significant performance degradation, several measures are typically employed:

• In sharded database environments, ensure local_msg_record uses the same sharding strategy as the user table to maintain local transactions and avoid cross-shard joins
• Add indexes on account_no and create_time fields
• Periodically clean up data in local_msg_record—this table doesn't need long retention periods; control single-table data volume

Scheduled Task Processing

After data is successfully written to the pending message table, introduce a scheduled dispatch program that periodically scans local_msg_record records and sends messages to MQ.

The scheduled task processing strategy involves three main steps:

1. Pull a batch of data from the database using pagination
2. Query the user table by account to construct the message body (account, registration time)
3. Send the message to the message server with retry mechanism

Task Execution Frequency

When planning this solution, clearly defining the scheduled task execution frequency is crucial.

Task frequency directly determines message-sending real-time performance. As the number of tasks requiring scheduling grows, most scheduling frameworks handle second-level scheduling poorly.调度 is typically minute-level, but minute-level scheduling introduces significant delays that most business requiremants cannot tolerate.

ElasticJob solves this through streaming task support. The approach: configure the task to dispatch at minute-level intervals, such as once per minute. Each dispatch queries data by pagination, processes a batch, then queries the next batch. If unprocessed data exists, continue until no more data remains before ending this execution cycle. If processing time exceeds one dispatch cycle, ElasticJob's misfire compensation mechanism triggers another dispatch.

During business peaks, this approach provides near real-time processing. Only during low business volumes, when no more unprocessed data exists after processing a batch, will newly arrived data face up to a 1-minute delay.

Implementation with ElasticJob

While RocketMQ provides transactional message mechanisms, many organizations use multiple types of message middleware, some without transactional message support. For architectural generality, we avoid depending on single middleware features.

Implement the solution using ElasticJob framework with key code examples.

Implement the DataflowJob interface for streaming tasks, which defines the core business logic:

public class CouponDispatchJob implements DataflowJob<PendingMessage> {
    private static final int BATCH_SIZE = 100;
    private static final String COUPON_TOPIC = "registration_promotion_topic";
    
    private IMessageDao messageDao;
    private MQProducer producer;
    
    @Override
    public List<PendingMessage> fetchData(ShardingContext context) {
        int totalShards = context.getShardingTotalCount();
        int currentShard = context.getShardingItem();
        int shardMod = currentShard % totalShards;
        
        return messageDao.findPendingMessages(shardMod, 0, BATCH_SIZE);
    }
    
    @Override
    public void processData(ShardingContext context, List<PendingMessage> records) {
        if (records == null || records.isEmpty()) {
            return;
        }
        
        for (PendingMessage msg : records) {
            String payload = serializeMessage(msg);
            String messageKey = msg.getAccountId();
            
            try {
                SendResult result = producer.send(
                    new Message(COUPON_TOPIC, null, messageKey, payload.getBytes(StandardCharsets.UTF_8))
                );
                
                msg.setMessageId(result.getMsgId());
                msg.setDeliveryStatus(1);
                msg.setLastModified(System.currentTimeMillis());
                messageDao.updateStatus(msg);
                
            } catch (MQException e) {
                handleFailure(msg);
            }
        }
    }
    
    private String serializeMessage(PendingMessage msg) {
        Map<String, Object> content = new HashMap<>();
        content.put("accountId", msg.getAccountId());
        content.put("registrationTime", msg.getCreateTimestamp());
        content.put("promotionType", msg.getCampaignType());
        return new ObjectMapper().writeValueAsString(content);
    }
    
    private void handleFailure(PendingMessage msg) {
        msg.setRetryAttempts(msg.getRetryAttempts() + 1);
        msg.setNextAvailableTime(System.currentTimeMillis() + RETRY_INTERVAL_MS);
        messageDao.updateRetryInfo(msg);
    }
}

FetchData Implementation

The fetchData method retrieves unprocessed data. ElasticJob calls fetchData on each task trigger to attempt data retreival. If this method returns data, ElasticJob invokes processData for business logic. After processing one batch, it calls fetchData again to check for more data. If unprocessed data exists, it continues calling processData until no more data is found.

Critical point: use ShardingContext to obtain task sharding information. ShardingTotalCount represents total shards, and ShardingItem represents the current shard number. Typically, use these with ID modulo operations for data distribution:

Total Shards: 4
Shard 0: Handles IDs where (id % 4 == 0)
Shard 1: Handles IDs where (id % 4 == 1)
Shard 2: Handles IDs where (id % 4 == 2)
Shard 3: Handles IDs where (id % 4 == 3)

ProcessData Implementation

The processData method handles business logic. Data retrieved via fetchData passes to processData for execution. In this example, the process involves assembling MQ messages from pending records, sending to the MQ server, updating pending records, and changing status from pending to sent.

Handling Partial Failures

A practical concern: if business logic sends to different MQ clusters based on message types, failures in one cluster can affect others. Consider this scenario:

If fetchData retrieves 3 messages per call, it might pull records with IDs 1, 2, 3 destined for cluster_a's topic_a. If cluster_a experiences an outage, these messages cannot be sent. Since these records aren't processed successfully in processData, their status doesn't update. Subsequent fetchData calls continue pulling the same records (1, 2, 3), blocking messages intended for cluster_b. This creates a critical business failure.

Solution: Retry Fields

Add two fields to the pending message table: current retry count (retry_count) and next minimum scheduling time (next_select_time).

When processing fails, increment the retry count and set the next minimum scheduling time. For example, adding one minute to current time ensures that during the next minute, the streaming task won't pull this data, allowing other records processing opportunities.

Task Configuration

Configure the task using Spring XML integration:

<job:dataflow id="CouponDispatchJob" 
              class="com.example.scheduler.CouponDispatchJob"
              registry-center-ref="zkRegistryCenter"
              cron="0 0/2 * * * ?"
              sharding-total-count="4"
              sharding-item-parameters="0=A,1=B,2=C,3=D"
              failover="true"
              streaming-process="true"/>

Configuration parameters:

• id: Unique task identifier
• class: Task implementation class
• registry-center-ref: ZooKeeper bean reference for ElasticJob coordinator
• cron: Cron expression for scheduling
• sharding-total-count: Total number of shards
• sharding-item-parameters: Shard parameters for each partition
• failover: Enable/disable automatic failover
• streaming-process: Enable streaming mode for continuous processing