当前位置：首页 > news >正文

海量聊天消息处理：ShardingJDBC分库分表、ClickHouse冷热数据分离、ES复合查询方案、Flink实时计算与SpringCloud集成

news 2025/6/8 8:28:33

海量聊天消息处理：ShardingJDBC分库分表、ClickHouse冷热数据分离、ES复合查询方案、Flink实时计算与SpringCloud集成

一、背景介绍

每天有2000万条聊天消息，一年下来几千万亿海量数据。为应对这种规模的数据存储和处理需求，本文将从以下几个方面提供解决方案：

使用ShardingJDBC技术进行合理的分库分表策略存放MySQL。
结合大数据同步ClickHouse实现冷热数据分离。
结合ElasticSearch提供多种复合查询方案。
结合Flink进行实时计算并提供代码示例。
结合SpringCloud给出集成方案并提供代码示例。

二、ShardingJDBC分库分表策略

1. 分库分表策略

假设我们有以下的业务场景：

每天新增2000万条聊天记录。
每张表存储约500万条数据。
每个数据库最多存储4张表。

基于此，我们可以设计如下的分库分表策略：

// 每个数据库包含4张表
// 数据库名：chat_db_0, chat_db_1, ..., chat_db_n
// 表名：chat_message_0, chat_message_1, ..., chat_message_3// 分库规则：根据用户ID的hash值对数据库数量取模
// 分表规则：根据时间戳对表数量取模

2. 建表语句

CREATE TABLE `chat_message` (`id` BIGINT(20) NOT NULL AUTO_INCREMENT COMMENT '主键',`user_id` BIGINT(20) NOT NULL COMMENT '用户ID',`message` VARCHAR(500) NOT NULL COMMENT '消息内容',`send_time` DATETIME NOT NULL COMMENT '发送时间',PRIMARY KEY (`id`),UNIQUE KEY `uk_user_send_time` (`user_id`, `send_time`),INDEX `idx_send_time` (`send_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='聊天消息表';// 分区配置
ALTER TABLE `chat_message`
PARTITION BY RANGE (YEAR(send_time)) (PARTITION p2023 VALUES LESS THAN (2024),PARTITION p2024 VALUES LESS THAN (2025),PARTITION p2025 VALUES LESS THAN (2026)
);

三、ClickHouse冷热数据分离

为了实现冷热数据分离，我们可以将最近7天内的数据定义为热数据，存储在内存中；超过7天的数据定义为冷数据，存储在磁盘中。

// ClickHouse建表语句
CREATE TABLE chat_message_clickhouse
(id UInt64,user_id UInt64,message String,send_time DateTime
) ENGINE = MergeTree()
ORDER BY (user_id, send_time)
PARTITION BY toYYYYMM(send_time);// 冷热数据分离策略
// 使用ReplicatedMergeTree引擎，将热数据存储在内存中，冷数据存储在磁盘上。

四、ES复合查询方案

以下是几种复合查询方案：

1. Bool Query

{"query": {"bool": {"must": [{ "match": { "message": "hello" } },{ "range": { "send_time": { "gte": "now-7d/d", "lte": "now/d" } } }]}}
}

2. Nested Query

{"query": {"nested": {"path": "user","query": {"bool": {"must": [{ "match": { "user.name": "John" } },{ "range": { "user.age": { "gte": 18 } } }]}}}}
}

五、Flink实时计算方案

1. 实时计算代码示例

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class ChatMessageProcessor {public static void main(String[] args) throws Exception {// 创建执行环境final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 读取聊天消息流DataStreaminputStream = env.socketTextStream("localhost", 9999);// 实时处理聊天消息DataStreamprocessedStream = inputStream.map(new MapFunction() {@Overridepublic String map(String value) throws Exception {return "Processed: " + value;}});// 输出处理结果processedStream.print();// 启动任务env.execute("Chat Message Processor");}
}

六、SpringCloud集成方案

1. SpringCloud集成代码示例

@SpringBootApplication
@EnableDiscoveryClient
public class ChatServiceApplication {public static void main(String[] args) {SpringApplication.run(ChatServiceApplication.class, args);}
}@RestController
@RequestMapping("/chat")
public class ChatController {@GetMapping("/message")public String getMessage() {return "Hello, this is a chat message!";}
}

七、Nacos配置方案

# application.yml
spring:cloud:nacos:discovery:server-addr: localhost:8848config:server-addr: localhost:8848file-extension: yaml

八、Maven依赖

org.apache.shardingspheresharding-jdbc-core4.1.1mysqlmysql-connector-java8.0.26ru.yandex.clickhouseclickhouse-jdbc0.3.2org.elasticsearch.clientelasticsearch-rest-high-level-client7.10.2org.apache.flinkflink-streaming-java_2.111.12.0org.springframework.cloudspring-cloud-starter-netflix-eureka-client2.2.5.RELEASEcom.alibaba.cloudspring-cloud-starter-alibaba-nacos-discovery2.2.3.RELEASE