如何扩大GA4 Measurement Protocol的每日事件发送量？

问题背景 #

我正在使用Measurement Protocol向Google Analytics 4（GA4）发送自定义事件，已知：

• Measurement Protocol支持在单个REST请求中发送同一用户的多个事件；
• 无法在单个REST请求中发送不同用户的多个事件。

业务场景涉及数百万用户，需执行数百万次REST请求，但当前循环发送请求时，数千次迭代后触发Google服务器限流，报错：

requests.exceptions.ConnectionError: HTTPSConnectionPool(host='www.google-analytics.com', port=443): Max retries exceeded with url: [...]

已实现退避重试策略（代码如下），但每日仅能发送约50万条事件，需要提升至百万级。参考GA4 Measurement Protocol官方文档。

当前实现代码 #

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

url = 'https://www.google-analytics.com/mp/collect/mp/collect?api_secret=YYYY&measurement_id=G-XXXX'

for request_data in all_requests:
    # Get payload for REST API request
    payload = create_payload(request_data)

    # Define retry-backoff strategy
    n_retries = 10
    backoff_factor = 10
    status_codes = [443]

    # Execute request using retry-backoff strategy
    session = requests.Session()
    retry = Retry(connect = n_retries, backoff_factor = backoff_factor, status_forcelist = status_codes)
    adapter = HTTPAdapter(max_retries = retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    response = session.post(f'{url}',
                            data=payload,
                            headers={'content-type': 'application/json'},
                            timeout=10)
    if response.status_code != requests.codes.no_content:
        raise SystemExit(f'Failed to send event with status code '
                        f'({response.status_code}) and parameters: {payload}')

扩容方案 #

1. 复用HTTP会话，减少连接开销 #

当前代码每次循环都创建新的requests.Session()，频繁建立/销毁TCP连接会大幅降低效率，还容易触发限流。将会话初始化逻辑移到循环外：

# 会话初始化移到循环外，复用连接
session = requests.Session()
# 修正状态码：443不是HTTP状态码，加入GA4限流/服务器错误常用码
retry = Retry(connect=10, backoff_factor=2, status_forcelist=[429, 500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
# 统一设置请求头，避免重复定义
session.headers.update({'content-type': 'application/json'})

url = 'https://www.google-analytics.com/mp/collect/mp/collect?api_secret=YYYY&measurement_id=G-XXXX'

for request_data in all_requests:
    payload = create_payload(request_data)
    try:
        response = session.post(url, data=payload, timeout=10)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        # 替换直接退出逻辑，改为记录错误后继续执行
        print(f"发送失败: {str(e)}, payload: {payload}")
        continue

2. 聚合同一用户的多事件，减少请求总量 #

利用Measurement Protocol的特性，提前按用户ID分组事件数据，将同一用户的多个事件打包到单个请求中发送：

from collections import defaultdict

# 先按用户ID分组所有事件数据
user_events = defaultdict(list)
for request_data in all_requests:
    user_id = request_data['user_id']  # 假设数据中包含用户标识字段
    user_events[user_id].append(request_data)

# 为每个用户生成包含多事件的payload
for user_id, events in user_events.items():
    # 构造多事件payload格式
    payload = {
        "user_id": user_id,  # 或client_id，根据你的用户标识方案
        "events": [create_event_payload(event) for event in events]
    }
    # 复用上述优化后的会话发送请求...

该方式能直接减少总请求数，降低触发限流的概率。

3. 异步并发请求，提升吞吐量 #

使用异步请求库aiohttp替代同步的requests，通过控制并发数同时发送多个请求：

import aiohttp
import asyncio

async def send_event(session, url, payload):
    try:
        async with session.post(url, json=payload, timeout=10) as response:
            response.raise_for_status()
            return True
    except Exception as e:
        print(f"发送失败: {str(e)}, payload: {payload}")
        return False

async def main():
    url = 'https://www.google-analytics.com/mp/collect/mp/collect?api_secret=YYYY&measurement_id=G-XXXX'
    # 设置信号量控制并发数，建议从30-50开始测试，避免瞬间请求过载
    semaphore = asyncio.Semaphore(50)
    
    async with aiohttp.ClientSession() as session:
        # 包装请求函数，加入并发限制
        async def bounded_send(payload):
            async with semaphore:
                return await send_event(session, url, payload)
        
        # 生成所有请求任务
        tasks = []
        for request_data in all_requests:
            payload = create_payload(request_data)
            tasks.append(bounded_send(payload))
        
        # 批量执行所有异步任务
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

4. 优化重试策略 #

• 降低backoff_factor：当前设置为10会导致退避间隔快速增大（10s→20s→40s...），调整为2-3可平衡重试频率与限流风险；
• 覆盖正确的重试状态码：加入429（请求限流）、5xx系列（服务器错误）；
• 动态调整重试逻辑：连续遇到429时，临时增大退避间隔或暂停请求几秒。

5. 清理无效请求，节省配额 #

• 确保payload完全符合GA4格式要求，必填字段（如client_id/user_id、events数组）无缺失；
• 通过event_id字段实现事件去重，避免重复发送；
• 过滤测试数据、无效数据，减少不必要的请求。

6. 申请提高官方配额 #

若以上优化仍无法满足需求，可通过Google Analytics官方支持渠道申请提高Measurement Protocol的请求配额。

内容的提问来源于stack exchange，提问作者dietrich