---
title: "食谱"
description: "具体的可复制粘贴方案——构建你自己的最小开发工具，将数据通过管道传递给 curl + jq，先回放历史记录，再切换到实时模式，并在消费端进行聚合。"
canonical_url: "https://evlog.zhcndoc.com/extend/consumer-recipes"
---
# 食谱
> 具体的可复制粘贴方案——构建你自己的最小开发工具，将数据通过管道传递给 curl + jq，先回放历史记录，再切换到实时模式，并在消费端进行聚合。
结合了 [进程内总线和流服务器](https://evlog.zhcndoc.com/extend/stream) 与 [文件系统读取器](https://evlog.zhcndoc.com/extend/fs-reader) 的真实世界模式。
启动一个本地开发工具或仪表盘，用来消费 evlog 宽事件。
- 选择来源：实时（通过 SSE 的流服务器）或历史记录（来自 `.evlog/logs` 的 `readFsLogs`），或两者兼用（先回放，再实时跟踪）
- 对于 SSE：通过 `.evlog/stream.url` 或 `GET /api/_evlog/stream-info` 发现 URL，绝不要硬编码端口
- 打开一个 `EventSource`，并将消息解码为 `{ evlog: '1', type, data }` 信封（`type` 为 `hello | event | replay | ping`）
- 对于运行在与开发服务器不同源上的浏览器标签页，通过流服务器的 `cors` 选项配置 CORS，并谨慎转发凭据
- 在消费端进行聚合（计数、延迟直方图、错误分组），并让服务器保持简单
- 在无服务器平台上跳过。该流在进程内运行
文档：[https://www.evlog.dev/extend/consumer-recipes](https://www.evlog.dev/extend/consumer-recipes)
## 1. 构建一个最小开发工具
实时事件面板本质上就是 `EventSource` + 列表。完整的线协议和发现规则（`.evlog/stream.url`、`/api/_evlog/stream-info`、`{ evlog: '1', type, data }` 信封以及身份验证）记录在[流页面](https://evlog.zhcndoc.com/extend/stream#wire-format)中。下面的每个方案都假定你已经通过其中一种机制获取了 URL。
### 纯 HTML + JS（直接放进任意页面）
```html
evlog 最小开发工具
```
保存为 `devtool.html`，在 evlog 已注入的开发服务器运行时，用任意浏览器标签页打开即可。这就是整个 MVP。
### Vue 3 组件
```vue
-
{{ e.level }}
{{ e.service }}
{{ e.action ?? e.message ?? e.path }}
```
### React 钩子
```ts
import { useEffect, useState } from 'react'
import type { WideEvent } from 'evlog'
export function useEvlogStream(url: string) {
const [events, setEvents] = useState([])
useEffect(() => {
if (!url) return
const es = new EventSource(url)
es.onmessage = (e) => {
const env = JSON.parse(e.data)
if (env.evlog !== '1') return
if (env.type === 'event' || env.type === 'replay') {
setEvents(prev => [env.data, ...prev].slice(0, 500))
}
}
return () => es.close()
}, [url])
return events
}
```
这就是完整的集成面。没有 SDK，也没有超出 `evlog` 导出的 `WideEvent` 之外的特殊类型。
## 2. 使用 curl + jq 进行快速 CLI 检查
URL 在 `.evlog/stream.url` 中：
```bash
URL=$(cat .evlog/stream.url)
curl -N "$URL" | jq -c 'select(.type == "event") | .data'
```
按需在客户端过滤：
```bash
# 仅错误
curl -sN "$URL" | jq -c 'select(.type == "event" and .data.level == "error") | .data'
# 仅某个服务
curl -sN "$URL" | jq -c 'select(.type == "event" and .data.service == "checkout") | .data'
# 慢请求
curl -sN "$URL" | jq -c 'select(.type == "event" and .data.durationMs > 500) | .data'
```
`-N` 让 `curl` 保持流式模式（不缓冲）。`-s` 表示静默。
## 3. 先回放历史，再切换到实时
磁盘历史（文件系统 drain）+ 来自流服务器的实时更新 = 从任意时间点都能获得完整视图。
```ts
import { readFsLogs } from 'evlog/fs'
import { readFile } from 'node:fs/promises'
import type { WideEvent } from 'evlog'
async function bootstrap(handle: (e: WideEvent) => void) {
// 1. 回放来自 `.evlog/logs/` 的最近一小时
const since = new Date(Date.now() - 60 * 60 * 1000)
for await (const event of readFsLogs({ since })) {
handle(event)
}
// 2. 切换到实时 SSE 流
const url = (await readFile('.evlog/stream.url', 'utf-8')).trim()
const es = new EventSource(url)
es.onmessage = (e) => {
const env = JSON.parse(e.data)
if (env.evlog !== '1') return
if (env.type === 'event' || env.type === 'replay') {
handle(env.data)
}
}
return () => es.close()
}
```
`readFsLogs` 会跳过日期范围之外的文件，因此即使你保留了数周历史，回放步骤也很快。若要仅尾随、不做磁盘回放，则改为用 `?since=` 访问流服务器，以复用进程内环形缓冲区。
## 4. Node / Bun 客户端（fetch + ReadableStream）
相同协议，无需 `EventSource` polyfill：
```ts
import { readFile } from 'node:fs/promises'
const url = (await readFile('.evlog/stream.url', 'utf-8')).trim()
const res = await fetch(url)
const reader = res.body!.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
const { value, done } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
let idx
while ((idx = buffer.indexOf('\n\n')) !== -1) {
const frame = buffer.slice(0, idx)
buffer = buffer.slice(idx + 2)
const dataLine = frame.split('\n').find(l => l.startsWith('data:'))
if (!dataLine) continue
const env = JSON.parse(dataLine.slice(5).trim())
if (env.type === 'event') console.log(env.data)
}
}
```
## 5. 在消费端进行过滤、转换、聚合
让服务器保持简单，并让每个消费者选择自己关心的内容：
```ts
// 仅错误
const errors = events.filter(e => e.level === 'error')
// 慢请求
const slowReqs = events.filter(e => (e.durationMs ?? 0) > 500)
// 按服务分组
const byService = Object.groupBy(events, e => e.service)
// 滚动错误率（最近 100 条事件）
const last100 = events.slice(0, 100)
const errorRate = last100.filter(e => e.level === 'error').length / last100.length
// 临时成本分析——之所以可行，是因为 evlog/ai 会在每次 AI 调用时写入 ai.* 字段
const totalCost = events
.filter(e => typeof e.ai?.estimatedCost === 'number')
.reduce((sum, e) => sum + (e.ai?.estimatedCost as number), 0)
```
## 6. 自托管的 "tail -f" 替代方案
如果消费者运行在同一台机器上，可以完全跳过网络：
```ts
import { tailFsLogs } from 'evlog/fs'
const ac = new AbortController()
process.on('SIGINT', () => ac.abort())
for await (const event of tailFsLogs({ signal: ac.signal })) {
if (event.level === 'error') notifyOps(event)
}
```
无需对正在运行的应用进行插桩即可工作，适合监视某个目录的 sidecar / observer 进程。
## 哪些情况会导致消费者失效
- **不要在 Vercel Functions / Cloudflare Workers / Lambda 上运行流服务器。** 每次调用都是一个独立的隔离环境；一个隔离环境中的订阅者永远看不到其他隔离环境发出的事件。对于跨实例扇出，请使用真正的 broker（Redis Streams、NATS、Pub/Sub）。
- **除非你的 evlog 配置会对数据进行脱敏，否则不要将涉及身份验证的数据放入宽事件中。** 服务器会原样转发应用发出的内容，包括任何未脱敏的 PII。
- **不要在服务器端进行过滤**（“只要错误事件”）。服务器的设计目标就是保持透明。在消费端进行过滤；这样一个过滤条件就不会让另一个消费者收不到事件。
以上每一点都是[流](https://evlog.zhcndoc.com/extend/stream)的属性，而不是消费者的属性：一个进程内总线、一个 SSE 桥接，以及一个会回放但永远不会持久化到磁盘的环形缓冲区。
## Sitemap
See the full [sitemap](https://evlog.zhcndoc.com/sitemap.md) for all pages.