MySQL频繁更新热点数据高并发场景下的具体解决方案

// 应用层累加，定期批量更新到数据库
public class HotSpotCounter {
    private Jedis jedis;
    
    public void increment(String key) {
        // 在Redis中累加
        jedis.incr(key);
    }
    
    // 定时任务，批量同步到数据库
    @Scheduled(fixedRate = 5000) // 每5秒同步一次
    public void syncToDatabase() {
        Set<String> keys = jedis.keys("counter:*");
        for (String key : keys) {
            Long value = Long.parseLong(jedis.get(key));
            String entityId = key.substring(8); // 去掉"counter:"前缀
            
            // 批量更新数据库
            updateDatabase(entityId, value);
            
            // 清空Redis计数器
            jedis.set(key, "0");
        }
    }
}

-- 原始热点表
CREATE TABLE page_views (
    page_id INT PRIMARY KEY,
    view_count BIGINT
);

-- 拆分为多行
CREATE TABLE page_views_sharded (
    page_id INT,
    shard_id INT,  -- 分片ID (0-15)
    view_count BIGINT,
    PRIMARY KEY (page_id, shard_id)
);

-- 更新时分散到不同分片
UPDATE page_views_sharded 
SET view_count = view_count + 1 
WHERE page_id = 1001 AND shard_id = RAND() * 16;

-- 查询时汇总
SELECT SUM(view_count) FROM page_views_sharded WHERE page_id = 1001;

@Component
public class ViewCountService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void recordView(int pageId) {
        // 发送到消息队列，异步处理
        kafkaTemplate.send("page-view-topic", String.valueOf(pageId));
    }
}

// 消费者，批量处理
@KafkaListener(topics = "page-view-topic")
public void batchUpdateViews(List<String> pageIds) {
    Map<Integer, Long> countMap = pageIds.stream()
        .collect(Collectors.groupingBy(Integer::parseInt, Collectors.counting()));
    
    // 批量更新数据库
    for (Map.Entry<Integer, Long> entry : countMap.entrySet()) {
        updatePageView(entry.getKey(), entry.getValue());
    }
}

-- 基于当前值的更新，减少锁竞争
UPDATE products 
SET stock = stock - 1 
WHERE id = 1001 AND stock > 0;

-- 或者使用版本号
UPDATE products 
SET stock = stock - 1, version = version + 1 
WHERE id = 1001 AND version = @current_version;

-- 对于计数类操作，使用READ-COMMITTED + 短事务
SET SESSION transaction_isolation = 'READ-COMMITTED';
BEGIN;
UPDATE counters SET value = value + 1 WHERE name = 'page_views';
COMMIT;

-- 写操作指向主库
UPDATE hot_table SET count = count + 1 WHERE id = 1; -- 主库

-- 读操作指向从库  
SELECT count FROM hot_table WHERE id = 1; -- 从库

public class RequestMerger {
    private Map<Integer, AtomicLong> counterMap = new ConcurrentHashMap<>();
    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public RequestMerger() {
        // 每100ms批量处理一次
        scheduler.scheduleAtFixedRate(this::flush, 100, 100, TimeUnit.MILLISECONDS);
    }
    
    public void increment(int id) {
        counterMap.computeIfAbsent(id, k -> new AtomicLong(0)).incrementAndGet();
    }
    
    private void flush() {
        Map<Integer, Long> snapshot = new HashMap<>();
        counterMap.forEach((id, atomic) -> {
            long value = atomic.getAndSet(0);
            if (value > 0) {
                snapshot.put(id, value);
            }
        });
        
        // 批量更新数据库
        batchUpdate(snapshot);
    }
}

-- 增加锁相关内存
SET GLOBAL innodb_buffer_pool_size = 8G;  -- 根据内存调整
SET GLOBAL innodb_log_file_size = 2G;     -- 增大日志文件
SET GLOBAL innodb_lock_wait_timeout = 10; -- 减少锁等待时间

-- 调整线程并发数
SET GLOBAL innodb_thread_concurrency = 0; -- 0表示不限制

// 使用 Redis Cluster 分散热点
public class DistributedCounter {
    public void increment(String key) {
        // 使用CRC32分片到不同的Redis节点
        int slot = CRC32.hash(key) % 16384;
        String redisNode = getNodeBySlot(slot);
        redisTemplate(redisNode).opsForValue().increment(key);
    }
}

-- 热点商品库存更新，秒杀时大量并发
UPDATE products SET stock = stock - 1 WHERE id = 1001 AND stock > 0;

-- 1. 库存分片
CREATE TABLE product_stock_shard (
    product_id INT,
    shard_id TINYINT, -- 0-9 10个分片
    stock INT,
    PRIMARY KEY (product_id, shard_id)
);

-- 2. 更新时随机选择分片
UPDATE product_stock_shard 
SET stock = stock - 1 
WHERE product_id = 1001 AND shard_id = FLOOR(RAND() * 10) AND stock > 0;

-- 3. 检查是否成功，如果失败重试其他分片

-- 查看行锁等待
SELECT * FROM information_schema.INNODB_LOCKS 
WHERE lock_table = 'your_hot_table';

-- 查看锁等待关系
SELECT * FROM information_schema.INNODB_LOCK_WAITS;

-- 查看InnoDB状态
SHOW ENGINE INNODB STATUS;

-- 查看当前运行的事务
SELECT * FROM information_schema.INNODB_TRX 
ORDER BY trx_started DESC 
LIMIT 10;