指标 · Cloudflare Queues docs

Queues 公开指标，允许您测量队列积压、消费者并发和消息操作。

Cloudflare 仪表板 ↗ 中显示的指标从 Cloudflare 的 GraphQL Analytics API 查询。您可以通过 GraphQL 或 HTTP 客户端以编程方式访问这些指标。

指标

积压

Queues 在 queuesBacklogAdaptiveGroups 数据集中导出以下指标。

指标	GraphQL 字段名	描述
积压字节数	`bytes`	积压的平均大小（字节）
积压消息数	`messages`	积压的平均大小（消息数）

queuesBacklogAdaptiveGroups 数据集提供以下维度用于过滤和分组查询：

queueID - 队列 ID
datetime - 消息发送时的时间戳
date - 消息发送时的时间戳，截断到天开始
datetimeHour - 消息发送时的时间戳，截断到小时开始
datetimeMinute - 消息发送时的时间戳，截断到分钟开始

消费者并发

Queues 在 queueConsumerMetricsAdaptiveGroups 数据集中导出以下指标。

指标	GraphQL 字段名	描述
平均消费者并发	`concurrency`	该时段内并发消费者的平均数量

queueConsumerMetricsAdaptiveGroups 数据集提供以下维度用于过滤和分组查询：

queueID - 队列 ID
datetime - 消费者指标的时间戳
date - 消费者指标的时间戳，截断到天开始
datetimeHour - 消费者指标的时间戳，截断到小时开始
datetimeMinute - 消费者指标的时间戳，截断到分钟开始

消息操作

Queues 在 queueMessageOperationsAdaptiveGroups 数据集中导出以下指标。

指标	GraphQL 字段名	描述
总可计费操作	`billableOperations`	该时段内可计费操作（写入、读取和删除）的总和
总字节数	`bytes`	从队列读取、写入和删除的字节总和
延迟	`lagTime`	消息写入与消费操作之间平均延迟时间（毫秒）。
重试	`retryCount`	每条消息的平均重试次数
消息大小	`messageSize`	指定时段内的最大消息大小

queueMessageOperationsAdaptiveGroups 数据集提供以下维度用于过滤和分组查询：

queueID - 队列 ID
actionType - 消息操作类型。可为 WriteMessage、ReadMessage 或 DeleteMessage
consumerType - 队列消费者类型。可为 worker 或 http。仅适用于 ReadMessage 和 DeleteMessage 操作类型
outcome - 消息操作结果。仅适用于 DeleteMessage 操作类型。可为 success、dlq 或 fail。
datetime - 消息操作的时间戳
date - 消息操作的时间戳，截断到天开始
datetimeHour - 消息操作的时间戳，截断到小时开始
datetimeMinute - 消息操作的时间戳，截断到分钟开始

实时积压

您可以通过 Queues REST API 和 JavaScript API 访问实时积压指标。这些指标提供时间点值而非聚合数据。

指标	字段名	描述
积压计数	`backlog_count`	队列中当前的消息数量
积压字节数	`backlog_bytes`	队列中消息的总大小（字节）
最旧消息时间戳	`oldest_message_timestamp_ms`	队列中最旧消息的时间戳（毫秒）

要通过 REST API 检索这些指标，请使用指标端点（/accounts/{account_id}/queues/{queue_id}/metrics）。

调用 JavaScript API 的 send()、sendBatch() 或 metrics() 时，这些字段也包含在 metadata.metrics 中。

GraphQL 查询示例

获取时段内平均队列积压

query QueueBacklog(
	$accountTag: string!
	$queueId: string!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			queueBacklogAdaptiveGroups(
				limit: 10000
				filter: {
					queueId: $queueId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
			) {
				avg {
					messages
					bytes
				}
			}
		}
	}
}

按小时获取平均消费者并发

query QueueConcurrencyByHour(
	$accountTag: string!
	$queueId: string!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			queueConsumerMetricsAdaptiveGroups(
				limit: 10000
				filter: {
					queueId: $queueId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
				orderBy: [datetimeHour_DESC]
			) {
				avg {
					concurrency
				}
				dimensions {
					datetimeHour
				}
			}
		}
	}
}

按分钟获取消息操作

query QueueMessageOperationsByMinute(
	$accountTag: string!
	$queueId: string!
	$datetimeStart: Date!
	$datetimeEnd: Date!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			queueMessageOperationsAdaptiveGroups(
				limit: 10000
				filter: {
					queueId: $queueId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
				orderBy: [datetimeMinute_DESC]
			) {
				count
				sum {
					bytes
				}
				dimensions {
					datetimeMinute
				}
			}
		}
	}
}