// 應(yīng)用層累加，定期批量更新到數(shù)據(jù)庫
public class HotSpotCounter {
    private Jedis jedis;
    
    public void increment(String key) {
        // 在Redis中累加
        jedis.incr(key);
    }
    
    // 定時任務(wù)，批量同步到數(shù)據(jù)庫
    @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:"前綴
            
            // 批量更新數(shù)據(jù)庫
            updateDatabase(entityId, value);
            
            // 清空Redis計數(shù)器
            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) {
        // 發(fā)送到消息隊列，異步處理
        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()));
    
    // 批量更新數(shù)據(jù)庫
    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;

-- 對于計數(shù)類操作，使用READ-COMMITTED + 短事務(wù)
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);
            }
        });
        
        // 批量更新數(shù)據(jù)庫
        batchUpdate(snapshot);
    }
}

-- 增加鎖相關(guān)內(nèi)存
SET GLOBAL innodb_buffer_pool_size = 8G;  -- 根據(jù)內(nèi)存調(diào)整
SET GLOBAL innodb_log_file_size = 2G;     -- 增大日志文件
SET GLOBAL innodb_lock_wait_timeout = 10; -- 減少鎖等待時間

-- 調(diào)整線程并發(fā)數(shù)
SET GLOBAL innodb_thread_concurrency = 0; -- 0表示不限制

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

-- 熱點商品庫存更新，秒殺時大量并發(fā)
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';

-- 查看鎖等待關(guān)系
SELECT * FROM information_schema.INNODB_LOCK_WAITS;

-- 查看InnoDB狀態(tài)
SHOW ENGINE INNODB STATUS;

-- 查看當前運行的事務(wù)
SELECT * FROM information_schema.INNODB_TRX 
ORDER BY trx_started DESC 
LIMIT 10;