前言

count 的常用方式，使用 count(*)来统计数据条数，但是 innodb 没有存储数据总数，所以执行起来就会很慢。

1. 可以使用 expalin sql 来返回预估行数，expalin select count(*)....., 通过预估的方式,统计数据条数。
2. 可以使用 redis 存储记录总数，或者一个额外的表来存储数据条数。

用了 Redis 来维持总数，那么就会涉及数据一致性的问题：

1. 如果数据短时间不一致但是业务可以接受的话，那么就可以考虑异步刷新 Redis 上的总数。

2. 使用 Canal 之类的工具监听 MySQL binlog，然后刷新 Redis 上的总数。

 这篇文章就来说说Canal

Canal的应用 

使用 canal 工具监听 MySQL binlog 并实时刷新 Redis 是一个常见的需求，特别是在需要将数据库变化同步到缓存系统时。以下是详细的步骤和相关代码示例来实现这一目标。 

1. 安装 Canal

Canal 是阿里巴巴开源的一个高性能 MySQL binlog 增量订阅 & 消费组件。首先，需要下载并安装 Canal。

安装步骤：

• 访问 Canal 官方 GitHub 仓库: GitHub - alibaba/canal: 阿里巴巴 MySQL binlog 增量订阅&消费组件
• 下载并解压 Canal Server。
• 修改 canal.properties 和 instance.properties 配置文件，根据你的 MySQL 配置进行修改。

2. 配置 Canal

在 instance.properties 文件中，设置 MySQL 连接信息，例如：

canal.instance.mysql.slaveId=1234
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=your_password
canal.instance.connectionCharset=UTF-8
canal.instance.defaultDatabaseName=test_db
canal.instance.filter.regex=test_db\\..*

3. 启动 Canal Server

在命令行中，进入 Canal 目录并启动 Canal Server：

sh bin/startup.sh

4. 编写 Canal Client

Canal 提供了 Java 客户端 API，用于消费 binlog 数据。你需要编写一个 Canal Client 来监听 binlog 并将变化的数据写入 Redis。

引入依赖

在你的 pom.xml 文件中添加 Canal 和 Redis 的依赖：

<dependencies>
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.5</version>
    </dependency>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>3.3.0</version>
    </dependency>
</dependencies>

编写 Canal Client 代码

以下是一个简单的 Canal Client 示例，它会将数据库变更同步到 Redis：

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import redis.clients.jedis.Jedis;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalClient {

    public static void main(String[] args) {
        // 创建 Canal 连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111), "example", "", "");

        try (Jedis jedis = new Jedis("localhost", 6379)) {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(100);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId != -1 && size > 0) {
                    processEntries(message.getEntries(), jedis);
                }
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    private static void processEntries(List<CanalEntry.Entry> entries, Jedis jedis) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                try {
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        if (eventType == CanalEntry.EventType.DELETE) {
                            handleDelete(rowData.getBeforeColumnsList(), jedis);
                        } else if (eventType == CanalEntry.EventType.INSERT) {
                            handleInsert(rowData.getAfterColumnsList(), jedis);
                        } else {
                            handleUpdate(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), jedis);
                        }
                    }
                } catch (Exception e) {
                    throw new RuntimeException("ERROR: " + entry.toString(), e);
                }
            }
        }
    }

    private static void handleInsert(List<CanalEntry.Column> columns, Jedis jedis) {
        // Example: Save the inserted row to Redis
        String key = null;
        for (CanalEntry.Column column : columns) {
            if ("id".equals(column.getName())) {
                key = "user:" + column.getValue();
                break;
            }
        }
        if (key != null) {
            for (CanalEntry.Column column : columns) {
                jedis.hset(key, column.getName(), column.getValue());
            }
        }
    }

    private static void handleDelete(List<CanalEntry.Column> columns, Jedis jedis) {
        // Example: Delete the row from Redis
        String key = null;
        for (CanalEntry.Column column : columns) {
            if ("id".equals(column.getName())) {
                key = "user:" + column.getValue();
                break;
            }
        }
        if (key != null) {
            jedis.del(key);
        }
    }

    private static void handleUpdate(List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns, Jedis jedis) {
        // Example: Update the row in Redis
        String key = null;
        for (CanalEntry.Column column : afterColumns) {
            if ("id".equals(column.getName())) {
                key = "user:" + column.getValue();
                break;
            }
        }
        if (key != null) {
            for (CanalEntry.Column column : afterColumns) {
                jedis.hset(key, column.getName(), column.getValue());
            }
        }
    }
}

5. 运行 Canal Client

编译并运行 Canal Client，它会监听 MySQL binlog 变更，并将变更实时同步到 Redis。

 

总结

通过上述步骤，你可以实现使用 Canal 工具监听 MySQL binlog，并实时刷新 Redis 的功能。这种方式可以确保数据库和缓存的一致性，适用于高性能、高实时性的数据同步场景。