本文探讨了在 Prometheus Python 客户端中，如何优雅且安全地从 `CollectorRegistry` 中获取已注册的指标对象（如 `Counter`）。我们将分析直接访问私有属性的局限性，并介绍两种主要解决方案：通过自定义类管理指标，以及通过子类化 `CollectorRegistry` 实现线程安全的指标获取方法，并推荐后者作为更专业的实践。
在 Prometheus Python 客户端中，CollectorRegistry 是用于存储和管理所有指标（如 Counter、Gauge、Histogram 等）的核心组件。然而，prometheus_client 库并未直接提供一个公共 API 来根据名称从注册表中检索已注册的指标对象。这使得在需要动态获取并操作现有指标时，开发者面临挑战。
通常，我们通过以下方式创建并注册一个 Counter：
from prometheus_client import CollectorRegistry, Counter, write_to_textfile
registry = CollectorRegistry()
metric_name = "my_application_requests_total"
documentation = "Total number of requests to the application."
counter = Counter(metric_name, documentation, registry=registry)
counter.inc(2) # 初始增加2
# 尝试获取并再次操作此 Counter
# 常见的非推荐做法是直接访问私有属性
# counter_retrieved = registry._names_to_collectors.get(metric_name)
# if isinstance(counter_retrieved, Counter):
# counter_retrieved.inc(3) # 再次增加3，总计为5
# write_to_textfile("counters.prom", registry)如上述代码注释所示，直接访问 registry._names_to_collectors 字典是许多开发者为了获取指标对象而采取的方式。然而，_names_to_collectors 是一个私有属性（以 _ 开头），这意味着它不属于公共 API，其内部结构和行为可能在未来的库版本中发生变化，导致代码兼容性问题。此外，直接访问私有属性也可能绕过库内部的线程安全机制，特别是在多线程环境中动态操作指标时，这可能导致数据不一致或竞态条件。
立即学习“Python免费学习笔记（深入）”；
库中提供了一个 restricted_registry 方法，但它主要用于生成一个只包含特定指标的子注册表，以便进行收集和导出，而不是用于获取原始指标对象以进行增量或观察操作。并且该方法被标记为“实验性”，且在处理带有标签的指标时，需要预先知道标签值，这限制了其在通用指标获取场景中的应用。
一种相对简单且快速的解决方案是创建一个自定义类来封装 CollectorRegistry，并在该类中维护一个指标对象的字典。当创建并注册新指标时，同时将其存储在这个自定义字典中，从而提供一个公共接口来获取这些指标。
from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram, Summary
class PrometheusMetricsManager:
"""
自定义指标管理类，封装 CollectorRegistry 并提供指标获取功能。
适用于指标在应用启动时一次性创建的场景。
"""
def __init__(self, registry: CollectorRegistry = None):
self._registry = registry if registry is not None else CollectorRegistry()
self._metrics = {} # 存储已注册的指标对象
def get_registry(self) -> CollectorRegistry:
return self._registry
def register_new_metrics(self, metrics_list: list):
"""
注册新的指标，并将其存储在内部字典中。
"""
for metric in metrics_list:
# 注册到 Prometheus 注册表
self._registry.register(metric)
# 存储到自定义字典，以指标名称为键
# 注意：对于有标签的指标，名称可能相同，但实例不同。
# 此处简化处理，以 metric.name 作为键，适用于无标签或名称唯一的场景。
# 对于有标签的复杂场景，可能需要更复杂的键生成策略。
self._metrics[metric._name] = metric # _name 是内部属性，但这里用于内部管理
def get_metric(self, metric_name: str):
"""
根据名称获取已注册的指标对象。
"""
return self._metrics.get(metric_name)
# 示例使用
if __name__ == "__main__":
metrics_manager = PrometheusMetricsManager()
# 创建并注册 Counter
app_requests_counter = Counter(
"app_requests_total",
"Total number of requests to the application.",
registry=metrics_manager.get_registry()
)
metrics_manager.register_new_metrics([app_requests_counter])
# 获取 Counter 并操作
retrieved_counter = metrics_manager.get_metric("app_requests_total")
if retrieved_counter:
retrieved_counter.inc(2) # 增加2
print(f"Current app_requests_total: {retrieved_counter._value}") # 访问私有_value仅用于演示
# 创建并注册 Gauge
cpu_usage_gauge = Gauge(
"cpu_usage_percent",
"Current CPU usage percentage.",
registry=metrics_manager.get_registry()
)
metrics_manager.register_new_metrics([cpu_usage_gauge])
retrieved_gauge = metrics_manager.get_metric("cpu_usage_percent")
if retrieved_gauge:
retrieved_gauge.set(75.5)
print(f"Current cpu_usage_percent: {retrieved_gauge._value}")
# 模拟原始问题中的操作
# 假设我们已经创建并注册了一个名为 "NAME" 的 Counter
original_counter = Counter("NAME", "DOCUMENTATION", registry=metrics_manager.get_registry())
metrics_manager.register_new_metrics([original_counter])
original_counter.inc(2) # 初始增加2
# 通过管理器获取并再次操作
retrieved_original_counter = metrics_manager.get_metric("NAME")
if isinstance(retrieved_original_counter, Counter):
retrieved_original_counter.inc(3) # 再次增加3，总计为5
print(f"Updated NAME_total: {retrieved_original_counter._value}")
# 将指标写入文件（如果需要）
write_to_textfile("managed_metrics.prom", metrics_manager.get_registry())注意事项：
更推荐且更健壮的解决方案是子类化 CollectorRegistry，并在自定义的子类中实现一个线程安全的指标获取方法。这种方法能够直接利用 CollectorRegistry 内部的线程锁 (self._lock)，从而保证在多线程环境下的操作安全。
from prometheus_client import CollectorRegistry, Counter, REGISTRY
import threading
class CustomRegistry(CollectorRegistry):
"""
子类化 CollectorRegistry，提供线程安全的指标获取方法。
"""
def __init__(self, auto_describe=False):
super().__init__(auto_describe=auto_describe)
# _lock 是父类 CollectorRegistry 的内部锁，我们在此处直接使用
# _names_to_collectors 也是父类的内部字典
def get_metric(self, metric_name: str):
"""
线程安全地根据名称获取已注册的指标对象。
"""
with self._lock: # 使用父类的线程锁确保操作的原子性
return self._names_to_collectors.get(metric_name)
# 示例使用
if __name__ == "__main__":
# 创建并注册自定义注册表实例
my_custom_registry = CustomRegistry()
# 可以选择将自定义注册表设置为默认注册表，或者单独使用
# REGISTRY.register(my_custom_registry) # 如果需要全局替换默认注册表
# 创建 Counter，并指定使用自定义注册表
name = "APP_METRIC_TOTAL"
app_counter = Counter(name, "Total count for APP_METRIC.", registry=my_custom_registry)
app_counter.inc(5)
# 通过自定义注册表获取 Counter 并操作
retrieved_app_counter = my_custom_registry.get_metric(name)
if isinstance(retrieved_app_counter, Counter):
retrieved_app_counter.inc(10) # 增加10，总计为15
print(f"Updated {name}: {retrieved_app_counter._value}")
# 模拟原始问题中的操作
# 假设我们已经创建并注册了一个名为 "NAME" 的 Counter
original_problem_name = "NAME"
problem_counter = Counter(original_problem_name, "DOCUMENTATION", registry=my_custom_registry)
problem_counter.inc(2) # 初始增加2
# 通过自定义注册表获取并再次操作
retrieved_problem_counter = my_custom_registry.get_metric(original_problem_name)
if isinstance(retrieved_problem_counter, Counter):
retrieved_problem_counter.inc(3) # 再次增加3，总计为5
print(f"Updated {original_problem_name}_total: {retrieved_problem_counter._value}")
# 将指标写入文件
write_to_textfile("custom_registry_metrics.prom", my_custom_registry)优点与最佳实践：
在 Prometheus Python 客户端中获取已注册的指标对象，应避免直接访问私有属性 _names_to_collectors。
根据你的具体需求和应用场景，选择最合适的方案。对于大多数需要动态获取和操作指标的生产环境应用，子类化 CollectorRegistry 是一个更可靠的选择。
以上就是Python Prometheus 客户端：获取已注册指标对象的最佳实践的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
656
收藏
