Menü
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/microwind/ai-skills --skill nosqlDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
Basierend auf der SOC-Berufsklassifikation
SKILL.md wird angezeigt
| name | NoSQL数据库应用 |
| description | 当设计NoSQL数据库时，选择合适的数据库类型，设计数据模型，优化查询性能。处理分布式事务，实现数据一致性，和扩展性设计。 |
| license | MIT |
NoSQL数据库是处理大规模数据、高并发访问、灵活数据结构的重要解决方案。不当的NoSQL设计会导致性能问题、数据不一致、扩展困难。
核心原则: 好的NoSQL设计应该数据模型合理、查询高效、扩展性强、一致性可控。坏的NoSQL设计会查询缓慢、数据冗余、扩展困难。
始终:
触发短语:
问题:
文档数据库中过度嵌套，导致查询性能差
错误示例:
- 文档深度超过5层
- 数组元素过多
- 频繁更新的嵌套字段
- 缺少合理的索引设计
解决方案:
1. 合理设计文档结构
2. 控制嵌套深度
3. 使用引用代替嵌套
4. 优化索引策略
问题:
NoSQL数据库查询性能差，响应时间长
错误示例:
- 缺少合适的索引
- 全表扫描查询
- 复杂聚合查询
- 不当的分页实现
解决方案:
1. 设计合理的索引
2. 优化查询语句
3. 使用聚合管道
4. 实现高效分页
问题:
分布式环境下数据一致性难以保证
错误示例:
- 缺少一致性策略
- 事务处理不当
- 缓存与数据库不一致
- 并发更新冲突
解决方案:
1. 选择合适的一致性级别
2. 实现分布式事务
3. 设计缓存更新策略
4. 处理并发冲突
问题:
数据库扩展困难，无法应对增长
错误示例:
- 分片策略不当
- 热点数据集中
- 跨分片查询过多
- 负载不均衡
解决方案:
1. 合理设计分片键
2. 实现数据分片
3. 优化跨分片查询
4. 实现负载均衡
import json
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import re
class DocumentType(Enum):
"""文档类型"""
USER = "user"
PRODUCT = "product"
ORDER = "order"
LOG = "log"
CONFIG = "config"
class IndexType(Enum):
"""索引类型"""
SINGLE = "single"
COMPOUND = "compound"
TEXT = "text"
HASHED = "hashed"
GEOSPATIAL = "geospatial"
@dataclass
class FieldDefinition:
"""字段定义"""
name: str
data_type: str
required: bool = False
index: bool = False
unique: bool = False
default: Any = None
validation: Dict[str, Any] = field(default_factory=dict)
:
name:
fields: []
index_type: IndexType
unique: =
sparse: =
expire_after: [] =
:
collection_name:
document_type: DocumentType
fields: [FieldDefinition]
indexes: [IndexDefinition] = field(default_factory=)
validation_rules: [, ] = field(default_factory=)
:
():
.schemas: [, DocumentSchema] = {}
.design_patterns = {
: ._design_embedding_pattern,
: ._design_referencing_pattern,
: ._design_bucket_pattern,
: ._design_schema_versioning_pattern,
}
() -> :
:
validation_result = ._validate_schema(schema)
validation_result[]:
()
.schemas[schema.collection_name] = schema
index_commands = ._generate_index_commands(schema)
validation_command = ._generate_validation_command(schema)
()
()
()
Exception e:
()
() -> DocumentSchema:
fields = [
FieldDefinition(
name=,
data_type=,
required=
),
FieldDefinition(
name=,
data_type=,
required=,
unique=,
index=,
validation={: , : }
),
FieldDefinition(
name=,
data_type=,
required=,
unique=,
index=,
validation={: }
),
FieldDefinition(
name=,
data_type=,
required=,
default={}
),
FieldDefinition(
name=,
data_type=,
required=,
default={}
),
FieldDefinition(
name=,
data_type=,
required=,
default=datetime.now
),
FieldDefinition(
name=,
data_type=,
required=,
default=datetime.now
),
FieldDefinition(
name=,
data_type=,
required=,
default=,
index=,
validation={: [, , ]}
)
]
indexes = [
IndexDefinition(
name=,
fields=[],
index_type=IndexType.SINGLE,
unique=
),
IndexDefinition(
name=,
fields=[],
index_type=IndexType.SINGLE,
unique=
),
IndexDefinition(
name=,
fields=[, ],
index_type=IndexType.COMPOUND
)
]
DocumentSchema(
collection_name=,
document_type=DocumentType.USER,
fields=fields,
indexes=indexes
)
() -> DocumentSchema:
fields = [
FieldDefinition(
name=,
data_type=,
required=
),
FieldDefinition(
name=,
data_type=,
required=,
unique=,
index=
),
FieldDefinition(
name=,
data_type=,
required=,
index=
),
FieldDefinition(
name=,
data_type=,
required=,
validation={: }
),
FieldDefinition(
name=,
data_type=,
required=,
validation={: }
),
FieldDefinition(
name=,
data_type=,
required=,
index=,
validation={: [, , , , ]}
),
FieldDefinition(
name=,
data_type=,
required=
),
FieldDefinition(
name=,
data_type=,
required=
),
FieldDefinition(
name=,
data_type=,
required=,
default=datetime.now
),
FieldDefinition(
name=,
data_type=,
required=,
default=datetime.now
)
]
indexes = [
IndexDefinition(
name=,
fields=[],
index_type=IndexType.SINGLE,
unique=
),
IndexDefinition(
name=,
fields=[],
index_type=IndexType.SINGLE
),
IndexDefinition(
name=,
fields=[, ],
index_type=IndexType.COMPOUND
),
IndexDefinition(
name=,
fields=[, ],
index_type=IndexType.COMPOUND
)
]
DocumentSchema(
collection_name=,
document_type=DocumentType.ORDER,
fields=fields,
indexes=indexes
)
() -> [, ]:
collection_name .schemas:
{: }
schema = .schemas[collection_name]
optimizations = []
pattern query_patterns:
optimization = ._analyze_query_pattern(schema, pattern)
optimizations.append(optimization)
recommendations = ._generate_optimization_recommendations(schema, optimizations)
{
: collection_name,
: query_patterns,
: optimizations,
: recommendations
}
() -> [, ]:
errors = []
warnings = []
schema.fields:
errors.append()
field_names = [field.name field schema.fields]
field_names:
warnings.append()
index schema.indexes:
field index.fields:
field field_names:
errors.append()
unique_fields = []
index schema.indexes:
index.unique (index.fields) == :
unique_fields.append(index.fields[])
field schema.fields:
field.unique field.name unique_fields:
warnings.append()
{
: (errors) == ,
: errors,
: warnings
}
() -> [[, ]]:
commands = []
index schema.indexes:
command = {
: schema.collection_name,
: [
{
: index.name,
: ._build_index_key(index),
: index.unique,
: index.sparse
}
]
}
index.expire_after:
command[][][] = index.expire_after
commands.append(command)
commands
() -> [, ]:
key = {}
field index.fields:
index.index_type == IndexType.TEXT:
key[field] =
index.index_type == IndexType.GEOSPATIAL:
key[field] =
index.index_type == IndexType.HASHED:
key[field] =
:
key[field] =
key
() -> :
schema.validation_rules:
validation_json = json.dumps(schema.validation_rules, ensure_ascii=)
() -> [, ]:
analysis = {
: pattern,
: []
}
pattern:
fields = ._extract_query_fields(pattern)
field fields:
field [f.name f schema.fields]:
has_index = (field index.fields index schema.indexes)
has_index:
analysis[].append()
analysis
() -> []:
fields = []
matches = re.findall(, query)
fields.extend(matches)
((fields))
() -> []:
recommendations = []
optimization optimizations:
recommendations.extend(optimization[])
recommendations = ((recommendations))
recommendations.sort()
recommendations
() -> [, ]:
{
: ,
: ,
: ,
: [, , ],
: [, , ]
}
() -> [, ]:
{
: ,
: ,
: ,
: [, , ],
: [, , ]
}
() -> [, ]:
{
: ,
: ,
: ,
: [, , ],
: [, , ]
}
() -> [, ]:
{
: ,
: ,
: ,
: [, , ],
: [, , ]
}
():
()
designer = MongoDBDocumentDesigner()
user_schema = designer.design_user_schema()
designer.create_schema(user_schema)
order_schema = designer.design_order_schema()
designer.create_schema(order_schema)
()
query_patterns = [
,
,
]
optimization = designer.optimize_document_structure(, query_patterns)
()
()
patterns = [
designer._design_embedding_pattern(, ),
designer._design_referencing_pattern(, ),
designer._design_bucket_pattern(),
designer._design_schema_versioning_pattern()
]
pattern patterns:
()
()
()
()
()
__name__ == :
main()
import json
import time
import hashlib
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass
from enum import Enum
import redis
from datetime import datetime, timedelta
class CacheStrategy(Enum):
"""缓存策略"""
LRU = "lru"
LFU = "lfu"
TTL = "ttl"
WRITE_THROUGH = "write_through"
WRITE_BEHIND = "write_behind"
WRITE_AROUND = "write_around"
class DataType(Enum):
"""数据类型"""
STRING = "string"
HASH = "hash"
LIST = "list"
SET = "set"
ZSET = "zset"
@dataclass
class CacheConfig:
"""缓存配置"""
key_prefix: str
default_ttl: int
max_memory: str
eviction_policy: str
data_type: DataType
compression: bool = False
serialization: str = "json"
@dataclass
class :
hits: =
misses: =
hit_rate: =
memory_usage: =
key_count: =
avg_ttl: =
:
():
.redis_client = redis.Redis(host=host, port=port, db=db, decode_responses=)
.configs: [, CacheConfig] = {}
.metrics: [, CacheMetrics] = {}
() -> :
:
.configs[config_name] = config
.metrics[config_name] = CacheMetrics()
.redis_client.config_set(, config.max_memory)
.redis_client.config_set(, config.eviction_policy)
()
Exception e:
()
() -> :
config_name .configs:
()
:
config = .configs[config_name]
full_key =
serialized_value = ._serialize_value(value, config.serialization)
config.compression:
serialized_value = ._compress_data(serialized_value)
ttl :
ttl = config.default_ttl
result = .redis_client.setex(full_key, ttl, serialized_value)
._update_metrics(config_name, )
result
Exception e:
()
() -> []:
config_name .configs:
()
:
config = .configs[config_name]
full_key =
cached_value = .redis_client.get(full_key)
cached_value :
._update_metrics(config_name, )
config.compression:
cached_value = ._decompress_data(cached_value)
value = ._deserialize_value(cached_value, config.serialization)
._update_metrics(config_name, )
value
Exception e:
()
._update_metrics(config_name, )
() -> :
config_name .configs:
()
:
config = .configs[config_name]
full_key =
result = .redis_client.delete(full_key)
._update_metrics(config_name, )
result >
Exception e:
()
() -> :
config_name .configs:
()
:
config = .configs[config_name]
full_pattern =
keys = .redis_client.keys(full_pattern)
keys:
deleted_count = .redis_client.delete(*keys)
._update_metrics(config_name, , deleted_count)
deleted_count
Exception e:
()
() -> [CacheMetrics]:
config_name .metrics:
metrics = .metrics[config_name]
total_requests = metrics.hits + metrics.misses
total_requests > :
metrics.hit_rate = metrics.hits / total_requests
:
info = .redis_client.info()
metrics.memory_usage = info.get(, )
metrics.key_count = info.get(, {}).get(, )
config = .configs[config_name]
keys = .redis_client.keys()
keys:
total_ttl =
count =
key keys[:]:
ttl = .redis_client.ttl(key)
ttl > :
total_ttl += ttl
count +=
count > :
metrics.avg_ttl = total_ttl / count
Exception e:
()
metrics
() -> [, ]:
config_name .configs:
{: }
results = {
: ,
: ,
: []
}
key keys:
:
data = data_loader(key)
data :
success = .set_cache(config_name, key, data)
success:
results[] +=
:
results[] +=
results[].append()
:
results[] +=
results[].append()
Exception e:
results[] +=
results[].append()
results
() -> :
cached_data = .get_cache(config_name, key)
cached_data :
cached_data
data = data_loader(key)
data :
.set_cache(config_name, key, data)
data
() -> :
serialization == :
json.dumps(value, ensure_ascii=, default=)
serialization == :
pickle
pickle.dumps(value).()
:
(value)
() -> :
serialization == :
json.loads(value)
serialization == :
pickle
pickle.loads(.fromhex(value))
:
value
() -> :
zlib
compressed = zlib.compress(data.encode())
compressed.()
() -> :
zlib
compressed = .fromhex(compressed_data)
decompressed = zlib.decompress(compressed)
decompressed.decode()
():
config_name .metrics:
metrics = .metrics[config_name]
operation == :
metrics.hits += count
operation == :
metrics.misses += count
operation == :
operation == :
operation == :
():
()
cache_manager = RedisCacheManager()
user_cache_config = CacheConfig(
key_prefix=,
default_ttl=,
max_memory=,
eviction_policy=,
data_type=DataType.HASH,
compression=,
serialization=
)
cache_manager.add_cache_config(, user_cache_config)
user_data = {
: ,
: ,
: ,
: {
: ,
:
}
}
cache_manager.set_cache(, , user_data)
cache_manager.set_cache(, , {: , : })
()
cached_user = cache_manager.get_cache(, )
()
cached_user2 = cache_manager.get_cache(, )
()
non_existent = cache_manager.get_cache(, )
()
()
metrics = cache_manager.get_cache_metrics()
metrics:
()
()
()
()
()
()
():
users = {
: {: , : , : },
: {: , : , : },
: {: , : , : }
}
users.get(user_id)
()
user_data = cache_manager.implement_cache_aside_pattern(, , load_user_data)
()
()
warmup_keys = [, , , , ]
warmup_results = cache_manager.implement_cache_warming(, load_user_data, warmup_keys)
()
()
invalidated_count = cache_manager.invalidate_pattern(, )
()
__name__ == :
main()