Canal解决select count(*)执行慢的问题-编程知识

Canal解决select count(*)执行慢的问题

前言

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

可以使用 expalin sql 来返回预估行数，expalin select count(*)....., 通过预估的方式,统计数据条数。
可以使用 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 RedisString 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 RedisString 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 RedisString 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());}}}
}