构建亿级实时仪表盘，毫秒级响应实战指南

Whoami：5年+金融、政府、医疗领域工作经验的DBA
Certificate：PGCM、OCP、YCP
Skill：Oracle、Mysql、PostgreSQL、国产数据库
Platform：CSDN、墨天轮、公众号（呆呆的私房菜）
复制

-- 分布式事件表设计
CREATE TABLE http_request (
  site_id INT,
  ingest_time TIMESTAMPTZ DEFAULT now(),
  url TEXT,
  request_country TEXT,
  ip_address TEXT,
  status_code INT,
  response_time_msec INT
);


-- 按site_id分片（建议分片数=CPU核心数x2~4）
SELECT create_distributed_table('http_request', 'site_id');
复制

DO $$
BEGIN
  LOOP
    INSERT INTO http_request (
      site_id, ingest_time, url,
      request_country, ip_address,
      status_code, response_time_msec
    ) VALUES (
      trunc(random() * 32), 
      clock_timestamp(),
      concat('http://example.com/', md5(random()::text)),
      (ARRAY['China','India','USA','Indonesia'])[ceil(random()*4)],
      concat(
        trunc(random()*250+2), '.',
        trunc(random()*250+2), '.',
        trunc(random()*250+2), '.',
        trunc(random()*250+2)
      )::inet,
      (ARRAY[200,404])[ceil(random()*2)],
      5+trunc(random()*150)
    );
    COMMIT;
    PERFORM pg_sleep(random() * 0.25);
  END LOOP;
END 
$$;
复制

SELECT 
  site_id,
  date_trunc('minute', ingest_time) as minute,
  COUNT(1) AS request_count,
  SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 1 ELSE 0 END) as success_count,
  SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 0 ELSE 1 END) as error_count,
  SUM(response_time_msec)/COUNT(1) AS average_response_time_msec 
FROM http_request
WHERE date_trunc('minute', ingest_time) > now() - INTERVAL '5 minutes'
GROUP BY site_id, minute
ORDER BY minute ASC;
复制

查询性能问题：全量扫描原始数据；
存储成本问题：原始数据长期存储成本高昂。
复制

CREATE TABLE http_request_1min (
  site_id INT,
  ingest_time TIMESTAMPTZ, -- 精确到分钟
  error_count INT,
  success_count INT,
  request_count INT,
  average_response_time_msec INT,
  CHECK (request_count = error_count + success_count),
  CHECK (ingest_time = date_trunc('minute', ingest_time))
);


SELECT create_distributed_table('http_request_1min', 'site_id');
CREATE INDEX http_request_1min_idx ON http_request_1min (site_id, ingest_time);
复制

相同的分片数（site_id 分片）
共置分片（相同 site_id 的原始数据和聚合数据存储在同一个节点）
复制

-- 创建定时汇总函数
-- 记录最后一次汇总时间
CREATE TABLE latest_rollup (
  minute timestamptz PRIMARY KEY CHECK (minute = date_trunc('minute', minute))
);
INSERT INTO latest_rollup VALUES ('1901-10-10'); -- 初始值


-- 创建汇总函数
CREATE OR REPLACE FUNCTION rollup_http_request() RETURNS void AS $$
DECLARE
  curr_rollup_time timestamptz := date_trunc('minute', now() - INTERVAL '1 minute');
  last_rollup_time timestamptz := (SELECT minute FROM latest_rollup);
BEGIN
  INSERT INTO http_request_1min (
    site_id, ingest_time, 
    request_count, success_count, 
    error_count, average_response_time_msec
  )
  SELECT 
    site_id,
    date_trunc('minute', ingest_time),
    COUNT(1),
    SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 1 ELSE 0 END),
    SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 0 ELSE 1 END),
    SUM(response_time_msec)/COUNT(1)
  FROM http_request
  WHERE ingest_time <@ tstzrange(last_rollup_time, curr_rollup_time, '(]')
  GROUP BY 1, 2;


  UPDATE latest_rollup SET minute = curr_rollup_time;
END;
$$ LANGUAGE plpgsql;
复制

# 系统 crontab 配置
* * * * * psql -c 'SELECT rollup_http_request();'
复制

SELECT 
  site_id,
  ingest_time AS minute,
  request_count,
  success_count,
  error_count,
  average_response_time_msec 
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - INTERVAL '5 minutes';
复制

-- 示例：保留最近1个月的原始数据
DELETE FROM http_request WHERE ingest_time < now() - INTERVAL '1 month';
复制