MongoDB分片技术实现
概述
MongoDB分片（Sharding）是MongoDB的水平扩展解决方案，通过将数据分布到多个分片（shard）上来处理大数据量和高吞吐量的需求。
MongoDB分片架构
1. 分片集群组件
# MongoDB分片集群架构
version: '3.8'
services:
# Config Server副本集
config1:
image: mongo:5.0
command: mongod --configsvr --replSet configReplSet --port 27019
ports:
- "27019:27019"
volumes:
- config1_data:/data/db
config2:
image: mongo:5.0
command: mongod --configsvr --replSet configReplSet --port 27019
ports:
- "27020:27019"
volumes:
- config2_data:/data/db
config3:
image: mongo:5.0
command: mongod --configsvr --replSet configReplSet --port 27019
ports:
- "27021:27019"
volumes:
- config3_data:/data/db
# 分片1副本集
shard1_replica1:
image: mongo:5.0
command: mongod --shardsvr --replSet shard1ReplSet --port 27018
ports:
- "27022:27018"
volumes:
- shard1_replica1_data:/data/db
shard1_replica2:
image: mongo:5.0
command: mongod --shardsvr --replSet shard1ReplSet --port 27018
ports:
- "27023:27018"
volumes:
- shard1_replica2_data:/data/db
# 分片2副本集
shard2_replica1:
image: mongo:5.0
command: mongod --shardsvr --replSet shard2ReplSet --port 27018
ports:
- "27024:27018"
volumes:
- shard2_replica1_data:/data/db
shard2_replica2:
image: mongo:5.0
command: mongod --shardsvr --replSet shard2ReplSet --port 27018
ports:
- "27025:27018"
volumes:
- shard2_replica2_data:/data/db
# mongos路由服务
mongos1:
image: mongo:5.0
command: mongos --configdb configReplSet/config1:27019,config2:27019,config3:27019 --port 27017
ports:
- "27017:27017"
depends_on:
- config1
- config2
- config3
mongos2:
image: mongo:5.0
command: mongos --configdb configReplSet/config1:27019,config2:27019,config3:27019 --port 27017
ports:
- "27026:27017"
depends_on:
- config1
- config2
- config3
volumes:
config1_data:
config2_data:
config3_data:
shard1_replica1_data:
shard1_replica2_data:
shard2_replica1_data:
shard2_replica2_data:2. 分片集群初始化脚本
#!/bin/bash
# mongodb-cluster-init.sh
echo "初始化MongoDB分片集群..."
# 等待服务启动
sleep 30
# 初始化Config Server副本集
echo "初始化Config Server副本集..."
mongo --host config1:27019 --eval '
rs.initiate({
_id: "configReplSet",
configsvr: true,
members: [
{ _id: 0, host: "config1:27019" },
{ _id: 1, host: "config2:27019" },
{ _id: 2, host: "config3:27019" }
]
})'
# 等待副本集初始化完成
sleep 20
# 初始化分片1副本集
echo "初始化分片1副本集..."
mongo --host shard1_replica1:27018 --eval '
rs.initiate({
_id: "shard1ReplSet",
members: [
{ _id: 0, host: "shard1_replica1:27018" },
{ _id: 1, host: "shard1_replica2:27018" }
]
})'
# 初始化分片2副本集
echo "初始化分片2副本集..."
mongo --host shard2_replica1:27018 --eval '
rs.initiate({
_id: "shard2ReplSet",
members: [
{ _id: 0, host: "shard2_replica1:27018" },
{ _id: 1, host: "shard2_replica2:27018" }
]
})'
# 等待分片副本集初始化完成
sleep 30
# 添加分片到集群
echo "添加分片到集群..."
mongo --host mongos1:27017 --eval '
sh.addShard("shard1ReplSet/shard1_replica1:27018,shard1_replica2:27018")
sh.addShard("shard2ReplSet/shard2_replica1:27018,shard2_replica2:27018")
'
echo "MongoDB分片集群初始化完成！"Java应用集成
1. Spring Boot配置
@Configuration
public class MongoShardingConfig {
@Value("${spring.data.mongodb.uri}")
private String mongoUri;
@Bean
public MongoClient mongoClient() {
// 连接到mongos路由服务
ConnectionString connectionString = new ConnectionString(mongoUri);
MongoClientSettings settings = MongoClientSettings.builder()
.applyConnectionString(connectionString)
.readPreference(ReadPreference.secondaryPreferred()) // 读写分离
.writeConcern(WriteConcern.MAJORITY) // 写关注
.readConcern(ReadConcern.MAJORITY) // 读关注
.retryWrites(true) // 重试写入
.retryReads(true) // 重试读取
.applyToConnectionPoolSettings(builder -> {
builder.maxSize(100) // 最大连接数
.minSize(10) // 最小连接数
.maxWaitTime(30, TimeUnit.SECONDS) // 最大等待时间
.maxConnectionIdleTime(60, TimeUnit.SECONDS); // 连接空闲时间
})
.build();
return MongoClients.create(settings);
}
@Bean
public MongoTemplate mongoTemplate() {
return new MongoTemplate(mongoClient(), "sharded_database");
}
}2. 分片键设计
@Document(collection = "users")
public class User {
@Id
private String id;
@Indexed
private String userId; // 分片键
private String username;
private String email;
private Date createTime;
private String region; // 地理位置
// 构造函数、getter、setter
}
@Document(collection = "orders")
public class Order {
@Id
private String id;
@Indexed
private String customerId; // 分片键
private String orderId;
private BigDecimal amount;
private Date orderTime;
private String status;
// 构造函数、getter、setter
}
@Document(collection = "products")
public class Product {
@Id
private String id;
@Indexed
private String categoryId; // 分片键
private String productName;
private BigDecimal price;
private String description;
// 构造函数、getter、setter
}3. 分片管理服务
@Service
public class MongoShardingService {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 启用数据库分片
*/
public void enableSharding(String database) {
Document command = new Document("enableSharding", database);
mongoTemplate.getDb().runCommand(command);
log.info("已启用数据库分片: {}", database);
}
/**
* 对集合进行分片
*/
public void shardCollection(String database, String collection, String shardKey) {
Document command = new Document("shardCollection", database + "." + collection)
.append("key", new Document(shardKey, 1));
mongoTemplate.getDb().runCommand(command);
log.info("已对集合进行分片: {}.{}, 分片键: {}", database, collection, shardKey);
}
/**
* 创建哈希分片
*/
public void createHashedSharding(String database, String collection, String shardKey) {
Document command = new Document("shardCollection", database + "." + collection)
.append("key", new Document(shardKey, "hashed"));
mongoTemplate.getDb().runCommand(command);
log.info("已创建哈希分片: {}.{}, 分片键: {}", database, collection, shardKey);
}
/**
* 创建范围分片
*/
public void createRangeSharding(String database, String collection, String shardKey) {
Document command = new Document("shardCollection", database + "." + collection)
.append("key", new Document(shardKey, 1));
mongoTemplate.getDb().runCommand(command);
log.info("已创建范围分片: {}.{}, 分片键: {}", database, collection, shardKey);
}
/**
* 创建复合分片键
*/
public void createCompoundSharding(String database, String collection,
Map<String, Object> shardKeys) {
Document keyDoc = new Document();
shardKeys.forEach(keyDoc::append);
Document command = new Document("shardCollection", database + "." + collection)
.append("key", keyDoc);
mongoTemplate.getDb().runCommand(command);
log.info("已创建复合分片: {}.{}, 分片键: {}", database, collection, shardKeys);
}
/**
* 查看分片状态
*/
public Document getShardingStatus() {
return mongoTemplate.getDb().runCommand(new Document("sh.status", 1));
}
/**
* 查看集合分片信息
*/
public Document getCollectionShardInfo(String database, String collection) {
Document command = new Document("collStats", collection)
.append("verbose", true);
return mongoTemplate.getDb(database).runCommand(command);
}
}4. 分片初始化配置
@Component
public class ShardingInitializer {
@Autowired
private MongoShardingService shardingService;
@EventListener(ApplicationReadyEvent.class)
public void initializeSharding() {
try {
// 启用数据库分片
shardingService.enableSharding("sharded_database");
// 用户集合 - 使用userId哈希分片
shardingService.createHashedSharding("sharded_database", "users", "userId");
// 订单集合 - 使用customerId范围分片
shardingService.createRangeSharding("sharded_database", "orders", "customerId");
// 产品集合 - 使用复合分片键
Map<String, Object> productShardKeys = new HashMap<>();
productShardKeys.put("categoryId", 1);
productShardKeys.put("productId", 1);
shardingService.createCompoundSharding("sharded_database", "products", productShardKeys);
log.info("MongoDB分片初始化完成");
} catch (Exception e) {
log.error("MongoDB分片初始化失败", e);
}
}
}分片策略优化
1. 智能分片键选择
@Service
public class ShardKeyOptimizer {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 分析集合的查询模式
*/
public ShardKeyRecommendation analyzeQueryPatterns(String collection) {
// 分析查询日志
List<Document> queryLogs = getQueryLogs(collection);
Map<String, Integer> fieldUsageCount = new HashMap<>();
Map<String, Double> fieldSelectivity = new HashMap<>();
for (Document log : queryLogs) {
Document query = log.get("command", Document.class);
if (query != null && query.containsKey("find")) {
Document filter = query.get("filter", Document.class);
if (filter != null) {
analyzeFilterFields(filter, fieldUsageCount);
}
}
}
// 计算字段选择性
for (String field : fieldUsageCount.keySet()) {
double selectivity = calculateFieldSelectivity(collection, field);
fieldSelectivity.put(field, selectivity);
}
return recommendShardKey(fieldUsageCount, fieldSelectivity);
}
private void analyzeFilterFields(Document filter, Map<String, Integer> fieldUsageCount) {
for (String field : filter.keySet()) {
fieldUsageCount.merge(field, 1, Integer::sum);
}
}
private double calculateFieldSelectivity(String collection, String field) {
// 计算字段的选择性（不重复值的比例）
Aggregation aggregation = Aggregation.newAggregation(
Aggregation.group(field),
Aggregation.count().as("distinctCount")
);
AggregationResults<Document> results = mongoTemplate.aggregate(
aggregation, collection, Document.class);
long distinctCount = results.getMappedResults().size();
long totalCount = mongoTemplate.count(new Query(), collection);
return totalCount > 0 ? (double) distinctCount / totalCount : 0;
}
private ShardKeyRecommendation recommendShardKey(Map<String, Integer> fieldUsageCount,
Map<String, Double> fieldSelectivity) {
// 综合考虑使用频率和选择性
String recommendedField = fieldUsageCount.entrySet().stream()
.max((e1, e2) -> {
double score1 = e1.getValue() * fieldSelectivity.getOrDefault(e1.getKey(), 0.0);
double score2 = e2.getValue() * fieldSelectivity.getOrDefault(e2.getKey(), 0.0);
return Double.compare(score1, score2);
})
.map(Map.Entry::getKey)
.orElse("_id");
return new ShardKeyRecommendation(recommendedField,
fieldSelectivity.getOrDefault(recommendedField, 0.0));
}
private List<Document> getQueryLogs(String collection) {
// 从MongoDB profiler获取查询日志
Query query = new Query(Criteria.where("ns").is("sharded_database." + collection)
.and("ts").gte(new Date(System.currentTimeMillis() - 24 * 60 * 60 * 1000))); // 最近24小时
return mongoTemplate.find(query, Document.class, "system.profile");
}
public static class ShardKeyRecommendation {
private String field;
private double selectivity;
public ShardKeyRecommendation(String field, double selectivity) {
this.field = field;
this.selectivity = selectivity;
}
// getter和setter
}
}2. 数据平衡监控
@Service
public class ShardBalanceMonitor {
@Autowired
private MongoTemplate mongoTemplate;
@Autowired
private MeterRegistry meterRegistry;
/**
* 监控分片数据分布
*/
@Scheduled(fixedRate = 300000) // 5分钟检查一次
public void monitorShardDistribution() {
try {
Document shardStats = mongoTemplate.getDb().runCommand(
new Document("shardDistribution", 1));
analyzeShardBalance(shardStats);
} catch (Exception e) {
log.error("分片分布监控失败", e);
}
}
private void analyzeShardBalance(Document shardStats) {
Document shards = shardStats.get("shards", Document.class);
if (shards == null) return;
Map<String, Long> shardSizes = new HashMap<>();
long totalSize = 0;
for (String shardName : shards.keySet()) {
Document shardInfo = shards.get(shardName, Document.class);
long size = shardInfo.getLong("size");
shardSizes.put(shardName, size);
totalSize += size;
}
// 计算分布不均衡度
double imbalanceRatio = calculateImbalanceRatio(shardSizes, totalSize);
// 记录指标
Gauge.builder("mongodb.shard.imbalance.ratio")
.register(meterRegistry, imbalanceRatio);
// 如果不均衡度超过阈值，触发重新平衡
if (imbalanceRatio > 0.3) { // 30%的不均衡度
log.warn("检测到分片数据不均衡，不均衡度: {:.2f}", imbalanceRatio);
triggerRebalance();
}
}
private double calculateImbalanceRatio(Map<String, Long> shardSizes, long totalSize) {
if (shardSizes.isEmpty() || totalSize == 0) return 0;
double avgSize = (double) totalSize / shardSizes.size();
double maxDeviation = shardSizes.values().stream()
.mapToDouble(size -> Math.abs(size - avgSize) / avgSize)
.max()
.orElse(0);
return maxDeviation;
}
private void triggerRebalance() {
try {
// 启动平衡器
mongoTemplate.getDb().runCommand(new Document("balancerStart", 1));
log.info("已启动分片重新平衡");
} catch (Exception e) {
log.error("启动分片重新平衡失败", e);
}
}
/**
* 监控chunk分布
*/
@Scheduled(fixedRate = 600000) // 10分钟检查一次
public void monitorChunkDistribution() {
try {
// 查询chunks集合
Query query = new Query();
List<Document> chunks = mongoTemplate.find(query, Document.class, "chunks");
Map<String, Integer> shardChunkCount = new HashMap<>();
for (Document chunk : chunks) {
String shard = chunk.getString("shard");
shardChunkCount.merge(shard, 1, Integer::sum);
}
// 记录每个分片的chunk数量
for (Map.Entry<String, Integer> entry : shardChunkCount.entrySet()) {
Gauge.builder("mongodb.shard.chunk.count")
.tag("shard", entry.getKey())
.register(meterRegistry, entry.getValue());
}
} catch (Exception e) {
log.error("Chunk分布监控失败", e);
}
}
}3. 查询路由优化
@Service
public class QueryRoutingOptimizer {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 优化查询以避免跨分片操作
*/
public <T> List<T> optimizedFind(Query query, Class<T> entityClass, String collection) {
// 分析查询是否包含分片键
if (containsShardKey(query, collection)) {
// 包含分片键，可以路由到特定分片
return mongoTemplate.find(query, entityClass, collection);
} else {
// 不包含分片键，需要广播查询
log.warn("查询不包含分片键，将执行跨分片查询: {}", query);
return mongoTemplate.find(query, entityClass, collection);
}
}
/**
* 批量查询优化
*/
public <T> List<T> optimizedBatchFind(List<String> shardKeyValues,
String shardKeyField,
Class<T> entityClass,
String collection) {
// 按分片键分组
Map<String, List<String>> shardGroups = groupByShardKey(shardKeyValues, shardKeyField);
List<T> results = new ArrayList<>();
// 并行查询各分片
shardGroups.entrySet().parallelStream().forEach(entry -> {
Query query = new Query(Criteria.where(shardKeyField).in(entry.getValue()));
List<T> shardResults = mongoTemplate.find(query, entityClass, collection);
synchronized (results) {
results.addAll(shardResults);
}
});
return results;
}
/**
* 聚合查询优化
*/
public <T> AggregationResults<T> optimizedAggregate(Aggregation aggregation,
String collection,
Class<T> outputType) {
// 检查聚合管道是否可以下推到分片
if (canPushDownToShards(aggregation)) {
return mongoTemplate.aggregate(aggregation, collection, outputType);
} else {
// 需要在mongos层进行聚合
log.warn("聚合操作需要在mongos层执行，可能影响性能");
return mongoTemplate.aggregate(aggregation, collection, outputType);
}
}
private boolean containsShardKey(Query query, String collection) {
// 获取集合的分片键信息
String shardKey = getShardKey(collection);
if (shardKey == null) return false;
// 检查查询条件是否包含分片键
Document queryDoc = query.getQueryObject();
return queryDoc.containsKey(shardKey);
}
private String getShardKey(String collection) {
try {
// 从config.collections获取分片键信息
Query query = new Query(Criteria.where("_id").is("sharded_database." + collection));
Document collectionInfo = mongoTemplate.findOne(query, Document.class, "collections");
if (collectionInfo != null) {
Document key = collectionInfo.get("key", Document.class);
if (key != null && !key.isEmpty()) {
return key.keySet().iterator().next();
}
}
} catch (Exception e) {
log.error("获取分片键失败", e);
}
return null;
}
private Map<String, List<String>> groupByShardKey(List<String> values, String shardKeyField) {
// 根据分片键值计算目标分片
Map<String, List<String>> groups = new HashMap<>();
for (String value : values) {
String targetShard = calculateTargetShard(value, shardKeyField);
groups.computeIfAbsent(targetShard, k -> new ArrayList<>()).add(value);
}
return groups;
}
private String calculateTargetShard(String shardKeyValue, String shardKeyField) {
// 简化的分片计算逻辑
int hash = shardKeyValue.hashCode();
int shardCount = getShardCount();
int shardIndex = Math.abs(hash) % shardCount;
return "shard" + shardIndex;
}
private int getShardCount() {
try {
Document listShards = mongoTemplate.getDb().runCommand(new Document("listShards", 1));
List<Document> shards = listShards.getList("shards", Document.class);
return shards != null ? shards.size() : 1;
} catch (Exception e) {
log.error("获取分片数量失败", e);
return 1;
}
}
private boolean canPushDownToShards(Aggregation aggregation) {
// 检查聚合管道是否包含可以下推到分片的操作
List<AggregationOperation> operations = aggregation.getOperations();
for (AggregationOperation operation : operations) {
if (operation instanceof GroupOperation ||
operation instanceof SortOperation ||
operation instanceof LimitOperation) {
// 这些操作通常需要在mongos层执行
return false;
}
}
return true;
}
}性能优化
1. 连接池优化
@Configuration
public class MongoConnectionOptimization {
@Bean
public MongoClientSettings mongoClientSettings() {
return MongoClientSettings.builder()
.applyToConnectionPoolSettings(builder -> {
builder.maxSize(200) // 最大连接数
.minSize(20) // 最小连接数
.maxWaitTime(30, TimeUnit.SECONDS) // 最大等待时间
.maxConnectionLifeTime(60, TimeUnit.MINUTES) // 连接最大生存时间
.maxConnectionIdleTime(30, TimeUnit.MINUTES) // 连接最大空闲时间
.maintenanceInitialDelay(0, TimeUnit.SECONDS)
.maintenanceFrequency(30, TimeUnit.SECONDS); // 维护频率
})
.applyToSocketSettings(builder -> {
builder.connectTimeout(10, TimeUnit.SECONDS) // 连接超时
.readTimeout(30, TimeUnit.SECONDS); // 读取超时
})
.applyToServerSettings(builder -> {
builder.heartbeatFrequency(10, TimeUnit.SECONDS) // 心跳频率
.minHeartbeatFrequency(500, TimeUnit.MILLISECONDS); // 最小心跳频率
})
.build();
}
}2. 批量操作优化
@Service
public class MongoBatchOptimization {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 批量插入优化
*/
public <T> void optimizedBatchInsert(List<T> documents, String collection) {
if (documents.isEmpty()) return;
// 按分片键分组
Map<String, List<T>> shardGroups = groupDocumentsByShardKey(documents, collection);
// 并行插入各分片
shardGroups.entrySet().parallelStream().forEach(entry -> {
List<T> shardDocuments = entry.getValue();
// 分批插入，避免单次操作过大
int batchSize = 1000;
for (int i = 0; i < shardDocuments.size(); i += batchSize) {
int endIndex = Math.min(i + batchSize, shardDocuments.size());
List<T> batch = shardDocuments.subList(i, endIndex);
mongoTemplate.insert(batch, collection);
}
});
}
/**
* 批量更新优化
*/
public void optimizedBatchUpdate(List<UpdateRequest> updateRequests, String collection) {
BulkOperations bulkOps = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, collection);
for (UpdateRequest request : updateRequests) {
bulkOps.updateOne(request.getQuery(), request.getUpdate());
}
BulkWriteResult result = bulkOps.execute();
log.info("批量更新完成，匹配: {}, 修改: {}",
result.getMatchedCount(), result.getModifiedCount());
}
/**
* 批量删除优化
*/
public void optimizedBatchDelete(List<Query> deleteQueries, String collection) {
BulkOperations bulkOps = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, collection);
for (Query query : deleteQueries) {
bulkOps.remove(query);
}
BulkWriteResult result = bulkOps.execute();
log.info("批量删除完成，删除数量: {}", result.getDeletedCount());
}
private <T> Map<String, List<T>> groupDocumentsByShardKey(List<T> documents, String collection) {
// 根据分片键对文档进行分组
Map<String, List<T>> groups = new HashMap<>();
String shardKey = getShardKey(collection);
if (shardKey == null) {
groups.put("default", documents);
return groups;
}
for (T document : documents) {
String shardKeyValue = extractShardKeyValue(document, shardKey);
String targetShard = calculateTargetShard(shardKeyValue);
groups.computeIfAbsent(targetShard, k -> new ArrayList<>()).add(document);
}
return groups;
}
private String extractShardKeyValue(Object document, String shardKey) {
// 使用反射或其他方式提取分片键值
try {
Field field = document.getClass().getDeclaredField(shardKey);
field.setAccessible(true);
Object value = field.get(document);
return value != null ? value.toString() : "";
} catch (Exception e) {
log.error("提取分片键值失败", e);
return "";
}
}
private String getShardKey(String collection) {
// 获取集合的分片键
return "userId"; // 简化实现
}
private String calculateTargetShard(String shardKeyValue) {
// 计算目标分片
return "shard0"; // 简化实现
}
public static class UpdateRequest {
private Query query;
private Update update;
public UpdateRequest(Query query, Update update) {
this.query = query;
this.update = update;
}
// getter和setter
}
}3. 索引优化
@Service
public class MongoIndexOptimization {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 创建分片友好的索引
*/
public void createShardFriendlyIndexes(String collection) {
// 1. 分片键索引（自动创建）
// 2. 复合索引（包含分片键）
Index compoundIndex = new Index()
.on("userId", Sort.Direction.ASC) // 分片键
.on("createTime", Sort.Direction.DESC)
.on("status", Sort.Direction.ASC);
mongoTemplate.indexOps(collection).ensureIndex(compoundIndex);
// 3. 查询优化索引
Index queryIndex = new Index()
.on("userId", Sort.Direction.ASC) // 分片键
.on("email", Sort.Direction.ASC)
.sparse(); // 稀疏索引
mongoTemplate.indexOps(collection).ensureIndex(queryIndex);
// 4. 地理位置索引
Index geoIndex = new Index()
.on("userId", Sort.Direction.ASC) // 分片键
.on("location", "2dsphere");
mongoTemplate.indexOps(collection).ensureIndex(geoIndex);
}
/**
* 监控索引使用情况
*/
@Scheduled(fixedRate = 3600000) // 1小时检查一次
public void monitorIndexUsage() {
List<String> collections = Arrays.asList("users", "orders", "products");
for (String collection : collections) {
try {
// 获取索引统计信息
Document indexStats = mongoTemplate.getDb()
.getCollection(collection)
.aggregate(Arrays.asList(
new Document("$indexStats", new Document())
))
.first();
if (indexStats != null) {
analyzeIndexUsage(collection, indexStats);
}
} catch (Exception e) {
log.error("监控索引使用情况失败: {}", collection, e);
}
}
}
private void analyzeIndexUsage(String collection, Document indexStats) {
Document accesses = indexStats.get("accesses", Document.class);
if (accesses != null) {
long ops = accesses.getLong("ops");
Date since = accesses.getDate("since");
if (ops == 0 && since != null) {
long daysSinceLastUse = (System.currentTimeMillis() - since.getTime()) / (24 * 60 * 60 * 1000);
if (daysSinceLastUse > 30) {
log.warn("索引 {} 在集合 {} 中超过30天未使用，考虑删除",
indexStats.getString("name"), collection);
}
}
}
}
/**
* 自动创建查询优化索引
*/
public void autoCreateQueryIndexes(String collection, List<Document> queryPatterns) {
Map<String, Integer> fieldUsageCount = new HashMap<>();
// 分析查询模式
for (Document query : queryPatterns) {
analyzeQueryFields(query, fieldUsageCount);
}
// 创建高频查询字段的索引
fieldUsageCount.entrySet().stream()
.filter(entry -> entry.getValue() > 100) // 使用次数超过100
.forEach(entry -> {
String field = entry.getKey();
if (!field.equals("_id")) { // 跳过默认索引
Index index = new Index().on(field, Sort.Direction.ASC);
mongoTemplate.indexOps(collection).ensureIndex(index);
log.info("为字段 {} 创建索引，使用频率: {}", field, entry.getValue());
}
});
}
private void analyzeQueryFields(Document query, Map<String, Integer> fieldUsageCount) {
for (String field : query.keySet()) {
if (!field.startsWith("$")) { // 跳过操作符
fieldUsageCount.merge(field, 1, Integer::sum);
}
}
}
}监控与运维
1. 分片集群监控
@Component
public class MongoShardingMonitor {
@Autowired
private MongoTemplate mongoTemplate;
@Autowired
private MeterRegistry meterRegistry;
/**
* 监控分片集群健康状态
*/
@Scheduled(fixedRate = 30000)
public void monitorClusterHealth() {
try {
// 检查mongos状态
Document isMaster = mongoTemplate.getDb().runCommand(new Document("isMaster", 1));
boolean isMongos = isMaster.getBoolean("ismaster", false);
// 检查分片状态
Document listShards = mongoTemplate.getDb().runCommand(new Document("listShards", 1));
List<Document> shards = listShards.getList("shards", Document.class);
int healthyShards = 0;
int totalShards = shards.size();
for (Document shard : shards) {
String state = shard.getString("state");
if ("1".equals(state)) {
healthyShards++;
}
}
// 记录指标
Gauge.builder("mongodb.cluster.shards.total")
.register(meterRegistry, totalShards);
Gauge.builder("mongodb.cluster.shards.healthy")
.register(meterRegistry, healthyShards);
// 检查平衡器状态
Document balancerStatus = mongoTemplate.getDb().runCommand(
new Document("balancerStatus", 1));
boolean balancerEnabled = balancerStatus.getBoolean("mode", false);
Gauge.builder("mongodb.cluster.balancer.enabled")
.register(meterRegistry, balancerEnabled ? 1 : 0);
} catch (Exception e) {
log.error("MongoDB集群健康监控失败", e);
}
}
/**
* 监控分片性能指标
*/
@Scheduled(fixedRate = 60000)
public void monitorShardPerformance() {
try {
Document serverStatus = mongoTemplate.getDb().runCommand(
new Document("serverStatus", 1));
// 连接数
Document connections = serverStatus.get("connections", Document.class);
if (connections != null) {
int current = connections.getInteger("current", 0);
int available = connections.getInteger("available", 0);
Gauge.builder("mongodb.connections.current")
.register(meterRegistry, current);
Gauge.builder("mongodb.connections.available")
.register(meterRegistry, available);
}
// 操作计数
Document opcounters = serverStatus.get("opcounters", Document.class);
if (opcounters != null) {
long insert = opcounters.getLong("insert");
long query = opcounters.getLong("query");
long update = opcounters.getLong("update");
long delete = opcounters.getLong("delete");
Counter.builder("mongodb.operations.insert")
.register(meterRegistry).increment(insert);
Counter.builder("mongodb.operations.query")
.register(meterRegistry).increment(query);
Counter.builder("mongodb.operations.update")
.register(meterRegistry).increment(update);
Counter.builder("mongodb.operations.delete")
.register(meterRegistry).increment(delete);
}
} catch (Exception e) {
log.error("MongoDB性能监控失败", e);
}
}
}2. 自动故障恢复
@Service
public class MongoFailoverService {
@Autowired
private MongoTemplate mongoTemplate;
@Autowired
private NotificationService notificationService;
/**
* 检测并处理分片故障
*/
@Scheduled(fixedRate = 15000)
public void detectAndHandleFailures() {
try {
Document listShards = mongoTemplate.getDb().runCommand(new Document("listShards", 1));
List<Document> shards = listShards.getList("shards", Document.class);
for (Document shard : shards) {
String shardId = shard.getString("_id");
String host = shard.getString("host");
String state = shard.getString("state");
if (!"1".equals(state)) {
handleShardFailure(shardId, host);
}
}
} catch (Exception e) {
log.error("分片故障检测失败", e);
}
}
private void handleShardFailure(String shardId, String host) {
log.error("检测到分片故障: {} ({})", shardId, host);
// 发送告警
notificationService.sendAlert(
"MongoDB分片故障",
String.format("分片 %s (%s) 发生故障，请及时处理", shardId, host)
);
// 尝试自动恢复
attemptAutoRecovery(shardId, host);
}
private void attemptAutoRecovery(String shardId, String host) {
try {
// 检查副本集状态
if (host.contains("/")) {
String[] parts = host.split("/");
String replSetName = parts[0];
String[] hosts = parts[1].split(",");
// 尝试连接副本集的其他成员
for (String memberHost : hosts) {
if (testConnection(memberHost)) {
log.info("副本集 {} 的成员 {} 仍然可用", replSetName, memberHost);
return;
}
}
}
// 如果所有成员都不可用，尝试重启服务
log.warn("分片 {} 的所有成员都不可用，需要手动干预", shardId);
} catch (Exception e) {
log.error("自动恢复失败", e);
}
}
private boolean testConnection(String host) {
try {
MongoClient testClient = MongoClients.create("mongodb://" + host);
testClient.getDatabase("admin").runCommand(new Document("ping", 1));
testClient.close();
return true;
} catch (Exception e) {
return false;
}
}
}总结
MongoDB分片技术是处理大规模数据的重要解决方案。成功实施需要考虑：
分片键设计：选择合适的分片键是关键，需要平衡查询性能和数据分布
架构规划：合理规划Config Server、分片和mongos的部署
查询优化：尽量包含分片键以避免跨分片查询
监控运维：建立完善的监控体系，及时发现和处理问题
最佳实践：
选择高基数、查询频繁的字段作为分片键
使用复合分片键提高查询效率
定期监控数据分布和性能指标
建立自动化的故障检测和恢复机制
通过合理的设计和实施，MongoDB分片可以为应用提供优秀的水平扩展能力。