net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 4096
net.ipv4.tcp_syncookies = 1
net.ipv4.tcp_tw_reuse = 1
net.core.netdev_max_backlog = 16384
net.ipv4.tcp_fin_timeout = 15
fs.file-max = 2000000
vm.overcommit_memory = 1
vm.swappiness = 1

# 监听与安全
bind 127.0.0.1 10.0.12.10           # 机房内网 IP
protected-mode yes
port 6379
# 如果走公网或跨机柜，开启 TLS（见下节）
# tls-port 6380

# 持久化：AOF 为主，RDB 作为 AOF 前导
appendonly yes
appendfilename "appendonly.aof"
appendfsync everysec                 # 性能与持久化折中
no-appendfsync-on-rewrite yes        # 避免重写期 fsync 抖动
auto-aof-rewrite-percentage 50
auto-aof-rewrite-min-size 256mb
aof-use-rdb-preamble yes

# 内存与过期策略（答题统计是热数据）
maxmemory 28gb
maxmemory-policy allkeys-lfu         # 热点友好，避免冷键撑爆
hash-max-ziplist-entries 1024
hash-max-ziplist-value 256

# 队列语义：用 Streams + 消费组
stream-node-max-bytes 4096
stream-node-max-entries 100

# 客户端与 I/O
maxclients 200000
timeout 0
tcp-keepalive 60
client-output-buffer-limit normal 0 0 0
io-threads 4                         # 7.x 开始支持（主要用于网络 I/O）

# 复制与故障切换
replica-read-only yes
repl-backlog-size 512mb

openssl req -x509 -nodes -newkey rsa:4096 -sha256 -days 3650 \
  -keyout /etc/ssl/private/redis.key \
  -out /etc/ssl/certs/redis.crt -subj "/CN=redis.internal"
cat /etc/ssl/certs/redis.crt /etc/ssl/private/redis.key \
  > /etc/ssl/private/redis.pem
chmod 600 /etc/ssl/private/redis.*

tls-port 6380
port 0
tls-cert-file /etc/ssl/private/redis.pem
tls-key-file  /etc/ssl/private/redis.key
tls-ca-cert-file /etc/ssl/certs/redis.crt

# app_producer.py
import time
import hashlib
import json
import redis
from fastapi import FastAPI


r = redis.Redis(host='10.0.12.10', port=6379, username='quizapp', password='p@ss-QuizApp')


app = FastAPI()
STREAM_PREFIX = "stream:answers:"


def sig(uid, qid, ts):
    return hashlib.sha256(f"{uid}:{qid}:{ts}".encode()).hexdigest()[:16]


@app.post("/submit")
def submit(quiz_id: str, uid: str, qid: str, ok: int):
    ts = int(time.time() * 1000)
    payload = {
        "uid": uid, "qid": qid, "ok": ok, "ts": ts, "sig": sig(uid, qid, ts)
    }
    sid = r.xadd(f"{STREAM_PREFIX}{quiz_id}", payload, maxlen=1000000, approximate=True)
    return {"stream_id": sid, "queued": True}

# worker_stats.py
import asyncio, time, os
import aioredis
from collections import defaultdict


STREAM = "stream:answers:{quiz}"
GROUP  = "group:stats"
CONSUMER = f"consumer-{os.uname().nodename}-{os.getpid()}"
BATCH = 200
ACK_AFTER = 0.5  # 秒，聚合窗口


async def ensure_group(r, quiz_id):
    try:
        await r.xgroup_create(STREAM.format(quiz=quiz_id), GROUP, id="$", mkstream=True)
    except aioredis.exceptions.ResponseError as e:
        if "BUSYGROUP" not in str(e): raise


async def bump(r, quiz_id, bucket):
    # 原子聚合：pipeline + HINCRBY；必要时用 Lua 做一次多键原子（示例下节）
    key = f"hash:quiz:{quiz_id}:stats"
    pipe = r.pipeline()
    for field, inc in bucket.items():
        pipe.hincrby(key, field, inc)
    await pipe.execute()


LUA_DEDUP = """
-- 幂等：用 SETNX 保证 sig 唯一消费，过期防膨胀
local k = KEYS[1]; local sig = ARGV[1]; local ttl = tonumber(ARGV[2])
if redis.call('SETNX', k .. sig, 1) == 1 then
  redis.call('PEXPIRE', k .. sig, ttl)
  return 1
else
  return 0
end
"""


async def consume_quiz(quiz_id):
    r = await aioredis.from_url("redis://10.0.12.10:6379", username="statsworker", password="p@ss-Worker")
    await ensure_group(r, quiz_id)
    dedup = r.register_script(LUA_DEDUP)
    buffer, last = defaultdict(int), time.time()


    while True:
        # 先处理 PEL（pending）中的超时消息
        # XAUTOCLAIM Redis 6.2+：把 idle>5s 的消息拉回本消费者
        try:
            _, claimed = await r.xautoclaim(STREAM.format(quiz=quiz_id), GROUP, CONSUMER, min_idle_time=5000, start_id="0-0", count=BATCH)
            for sid, fields in claimed:
                # 幂等判断
                ok = await dedup(keys=[f"dedup:quiz:{quiz_id}:"], args=[fields[b"sig"].decode(), 60000])
                if ok == 1:
                    qid = fields[b"qid"].decode(); u = fields[b"uid"].decode()
                    isok = int(fields[b"ok"].decode())
                    buffer["total"] += 1
                    buffer[f"q:{qid}:total"] += 1
                    if isok: 
                        buffer["correct"] += 1
                        buffer[f"q:{qid}:correct"] += 1
                        buffer[f"u:{u}:streak"] += 1
                    else:
                        buffer[f"u:{u}:streak"] = 0
                await r.xack(STREAM.format(quiz=quiz_id), GROUP, sid)
        except Exception:
            pass


        # 常规读取新消息
        resp = await r.xreadgroup(GROUP, CONSUMER, streams={STREAM.format(quiz=quiz_id): ">"}, count=BATCH, latest_ids=None, timeout=200)
        if resp:
            for _stream, messages in resp:
                for sid, fields in messages:
                    ok = await dedup(keys=[f"dedup:quiz:{quiz_id}:"], args=[fields[b"sig"].decode(), 60000])
                    if ok == 1:
                        qid = fields[b"qid"].decode(); u = fields[b"uid"].decode()
                        isok = int(fields[b"ok"].decode())
                        buffer["total"] += 1
                        buffer[f"q:{qid}:total"] += 1
                        if isok:
                            buffer["correct"] += 1
                            buffer[f"q:{qid}:correct"] += 1
                            buffer[f"u:{u}:streak"] += 1
                        else:
                            buffer[f"u:{u}:streak"] = 0
                    await r.xack(STREAM.format(quiz=quiz_id), GROUP, sid)


        # 周期性 flush 到 Redis 哈希（再由另一个任务批量落库）
        if time.time() - last > ACK_AFTER and buffer:
            await bump(r, quiz_id, buffer)
            buffer.clear()
            last = time.time()


if __name__ == "__main__":
    asyncio.run(consume_quiz("spring_festival_2025"))

-- KEYS[1] = dedup prefix, KEYS[2] = stats key
-- ARGV = sig, ttl, qid, uid, ok
local dpx = KEYS[1]; local stats = KEYS[2]
local sig = ARGV[1]; local ttl = tonumber(ARGV[2])
local qid = ARGV[3]; local uid = ARGV[4]; local ok = tonumber(ARGV[5])


if redis.call('SETNX', dpx .. sig, 1) == 0 then
  return 0
end
redis.call('PEXPIRE', dpx .. sig, ttl)


redis.call('HINCRBY', stats, 'total', 1)
redis.call('HINCRBY', stats, 'q:'..qid..':total', 1)
if ok == 1 then
  redis.call('HINCRBY', stats, 'correct', 1)
  redis.call('HINCRBY', stats, 'q:'..qid..':correct', 1)
  redis.call('HINCRBY', stats, 'u:'..uid..':streak', 1)
else
  redis.call('HSET', stats, 'u:'..uid..':streak', 0)
end
return 1

[Unit]
Description=Quiz Stats Worker
After=network.target redis-server.service


[Service]
User=www-data
Environment=PYTHONUNBUFFERED=1
ExecStart=/usr/bin/python3 /srv/quiz/worker_stats.py
Restart=always
RestartSec=1
LimitNOFILE=1000000


[Install]
WantedBy=multi-user.target

port 26379
sentinel monitor quiz-redis 10.0.12.10 6379 2
sentinel down-after-milliseconds quiz-redis 5000
sentinel failover-timeout quiz-redis 60000
sentinel parallel-syncs quiz-redis 1

# 观察流长度与消费
redis-cli XINFO STREAM stream:answers:xxx
redis-cli XPENDING stream:answers:xxx group:stats

# 手动重分配 pending
redis-cli XAUTOCLAIM stream:answers:xxx group:stats consumer-1 5000 0-0 COUNT 100

# AOF 体检
redis-cli INFO persistence | egrep 'aof_current_size|aof_last_bgrewrite_status'

# 慢日志
redis-cli CONFIG SET slowlog-log-slower-than 1000
redis-cli SLOWLOG GET 10