CF Worker 101

边缘计算，更快触达

moz

Mar 11, 2025

序

CF是cloudflare的简称，后文皆用CF代指

一、概念

什么是CF worker？

CF用自己的边缘服务器提供Javascript运行的方式

适合构建API、反向代理

不适合fs操作，

不适合长时间运行的任务，单次请求CPU时间为免费10ms，付费30ms，

二、创建项目

pnpm create cloudflare

wrangler: 命令行工具，用于管理works和pages应用、D1数据库、KV存储、R2文件存储

// 全局安装

pnpm i wrangler -g

配置环境变量

// 定义类型
export interface Env {
  
}


// 环境变量

[[kv_namespaces]]
binding = ""
id=""

三、使用场景

定期执行代码，用于定时报告、监控警告

// schedule

export default {
  async scheduled(event, env, ctx) {
    ctx.waitUntil(doSomeTaskOnASchedule());
  },
};

// wrangler.toml

[triggers]
crons = ["0 0 * * *"] # 每天午夜运行

发送请求

// fetch

export default {
  async fetch(request, env, ctx) {
    return new Response('Hello World!');
  },
};

通过Durable Objects实现的警报功能，允许设置延迟执行的任务，比如订单超时处理

// alarm

export class AlarmHandler {
  constructor(state, env) {
    this.state = state;
    this.env = env;
    this.storage = state.storage;
  }
  
  async fetch(request) {
    // 解析请求
    const url = new URL(request.url);
    const path = url.pathname;
    
    // 处理不同的路径
    if (path === '/set-alarm' && request.method === 'POST') {
      return this.setAlarm(request);
    } else if (path === '/alarm-triggered') {
      return this.handleAlarmTriggered();
    }
    
    return new Response('Not Found', { status: 404 });
  }
  
  async setAlarm(request) {
    const { delaySeconds, data } = await request.json();
    
    if (!delaySeconds || delaySeconds <= 0) {
      return new Response('Invalid delay time', { status: 400 });
    }
    
    // 存储要在警报触发时使用的数据
    await this.storage.put('alarmData', data);
    
    // 设置闹钟
    const alarmId = this.state.storage.setAlarm(
      Date.now() + delaySeconds * 1000,
      new Request('https://internal/alarm-triggered')
    );
    
    return new Response(JSON.stringify({ 
      success: true, 
      message: `Alarm set to trigger in ${delaySeconds} seconds`,
      alarmId: alarmId
    }), {
      headers: { 'Content-Type': 'application/json' }
    });
  }
  
  async handleAlarmTriggered() {
    // 获取存储的数据
    const data = await this.storage.get('alarmData');
    
    // 执行警报任务
    console.log('Alarm triggered with data:', data);
    
    // 例如：发送通知
    if (data && data.notifyUrl) {
      await fetch(data.notifyUrl, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          message: 'Alarm triggered',
          timestamp: new Date().toISOString(),
          data: data
        })
      });
    }
    
    // 清理数据
    await this.storage.delete('alarmData');
    
    return new Response('Alarm handled successfully');
  }
  
  // 处理警报事件
  async alarm() {
    // 获取存储的数据
    const data = await this.storage.get('alarmData');
    
    // 执行警报任务
    console.log('Alarm event triggered with data:', data);
    
    // 执行相同的通知逻辑
    if (data && data.notifyUrl) {
      await fetch(data.notifyUrl, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          message: 'Alarm triggered',
          timestamp: new Date().toISOString(),
          data: data
        })
      });
    }
    
    // 清理数据
    await this.storage.delete('alarmData');
  }
}

// 主Worker脚本
export default {
  async fetch(request, env) {
    const url = new URL(request.url);
    
    if (url.pathname === '/schedule-task' && request.method === 'POST') {
      const { taskId, delaySeconds, data } = await request.json();
      
      // 创建或获取Durable Object
      const id = env.ALARM_HANDLER.idFromName(taskId || 'default');
      const alarmObj = env.ALARM_HANDLER.get(id);
      
      // 转发请求到Durable Object
      return await alarmObj.fetch(new Request('https://internal/set-alarm', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ delaySeconds, data })
      }));
    }
    
    return new Response('Not Found', { status: 404 });
  }
};

邮件自定义处理、过滤和分类、自动回复、webhook转发

// email

// 从Web请求触发邮件发送
export default {
  async fetch(request, env) {
    if (request.method === 'POST' && new URL(request.url).pathname === '/send-email') {
      try {
        // 从请求体获取邮件数据
        const { to, subject, body } = await request.json();
        
        // 基本验证
        if (!to || !subject || !body) {
          return new Response('Missing required fields', { status: 400 });
        }
        
        // 发送邮件
        await env.NOTIFICATION_EMAIL.send({
          to: to,
          from: '[email protected]',
          subject: subject,
          text: body,
          html: body.replace(/\n/g, '<br>')
        });
        
        return new Response('Email sent successfully', { status: 200 });
      } catch (error) {
        return new Response(`Error sending email: ${error.message}`, { status: 500 });
      }
    }
    
    return new Response('Method not allowed', { status: 405 });
  }
};

// wrangler.toml

[email_bindings]
binding_name = "NOTIFICATION_EMAIL"

消息队列,用于批处理、异步任务、日志收集

// 生产者

// 发送消息到队列的Worker
export default {
  async fetch(request, env) {
    // 创建要发送的消息
    const message = {
      url: request.url,
      method: request.method,
      timestamp: Date.now(),
      userId: request.headers.get('X-User-Id') || 'anonymous'
    };
    
    // 发送消息到队列
    await env.MY_QUEUE.send(message);
    
    return new Response('Request logged', { status: 200 });
  }
};

//消费者
// 处理队列消息的Worker
export default {
  // 常规请求处理
  async fetch(request, env) {
    return new Response('Queue consumer worker');
  },
  
  // 队列消息处理
  async queue(batch, env) {
    // 批量处理多条消息
    const results = [];
    
    for (const message of batch.messages) {
      try {
        // 解析消息内容
        const data = message.body;
        
        // 处理消息
        console.log(`Processing message: ${JSON.stringify(data)}`);
        
        // 执行某些操作，如存储到KV、发送到外部API等
        await processMessage(data, env);
        
        // 记录成功
        results.push({
          messageId: message.id,
          status: 'success'
        });
      } catch (error) {
        // 记录错误
        console.error(`Failed to process message ${message.id}: ${error}`);
        results.push({
          messageId: message.id,
          status: 'error',
          error: error.message
        });
      }
    }
    
    return { results };
  }
};

async function processMessage(data, env) {
  // 例如：存储访问日志
  const key = `log:${data.timestamp}:${crypto.randomUUID()}`;
  await env.LOGS_KV.put(key, JSON.stringify(data));
  
  // 如果需要，可以将处理后的数据发送到其他服务
  if (data.userId !== 'anonymous') {
    await fetch('https://analytics.example.com/track', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(data)
    });
  }
}

// wrangler.toml

[[queues.producers]]
queue = "my-queue"
binding = "MY_QUEUE"

[[queues.consumers]]
queue = "my-queue"
max_batch_size = 10
max_batch_timeout = 5

允许实时监控和记录Worker的日志和请求详情，比如调试与故障排除、性能监控、流量分析

// tail

export default {
  async tail(events, env, ctx) {
    fetch("<YOUR_ENDPOINT>", {
      method: "POST",
      body: JSON.stringify(events),
    })
  }
}

四、使用框架，提升效率

hono：运行在edge runtime的简单、超快的web框架

pnpm i hono

moz’s 订阅

Discussion about this post