同步收件人记录

使用 Email Sending 事件订阅，在出现投递问题后更新应用记录。本示例使用 Cloudflare Queues 与 Workers KV，从事务性通知中移除收件人。

准备资源

开始之前：

将每个符合条件的收件人地址作为键存储在 KV 中。值可以包含通知偏好或相关元数据。

查看事件流程

Email Sending 发布 bounce 与 complaint 事件。
队列将这些事件投递给 Worker。
Worker 从 KV 中移除不符合条件的收件人记录。

选择移除事件

对每个 message.complained 事件都移除记录。这些事件表示收件人将邮件报告为垃圾邮件。

仅当 payload.bounce.type 为 "hard" 时，才移除 bounce 记录。临时失败会在仍有重试时产生 message.deferred 事件。用尽临时重试后，可能产生 bounce 类型为 "soft" 的 message.bounced 事件。

有关 payload 详情，请参阅可用的 Email Sending 事件。

创建队列与订阅

创建队列并订阅到你的发送域名：

在 Cloudflare 仪表板中，前往 Queues 页面。创建一个名为 email-events 的队列。
Go to Queues ↗
选择 email-events，然后选择 Subscriptions（订阅） > Subscribe to events（订阅事件）。
输入订阅名称，并选择 Email Sending（电子邮件发送） 作为源。
选择你的发送域名，以及 message.bounced 和 message.complained 事件。
选择 Subscribe（订阅）。

配置 Worker

绑定 KV namespace，并将 Worker 注册为队列消费者：

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "recipient-record-sync",
  "main": "src/index.ts",
  // Set this to today's date
  "compatibility_date": "2026-08-17",
  "kv_namespaces": [
    {
      "binding": "RECIPIENTS",
      "id": "<RECIPIENTS_KV_NAMESPACE_ID>"
    }
  ],
  "queues": {
    "consumers": [
      {
        "queue": "email-events",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "email-events-dlq"
      }
    ]
  }
}

name = "recipient-record-sync"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-17"

[[kv_namespaces]]
binding = "RECIPIENTS"
id = "<RECIPIENTS_KV_NAMESPACE_ID>"

[[queues.consumers]]
queue = "email-events"
max_batch_size = 10
max_retries = 3
dead_letter_queue = "email-events-dlq"

该配置会在部署时创建 email-events-dlq。三次重试后，Queues 会将事件移至该死信队列。

添加队列消费者

queue() 处理程序会独立处理每个事件。它会删除适用的收件人记录，并对失败的 KV 操作进行重试。

src/index.jsjs

export default {
	async queue(batch, env) {
		for (const message of batch.messages) {
			try {
				const event = message.body;

				if (shouldRemove(event)) {
					await removeRecipient(env, event);
				}

				message.ack();
			} catch (error) {
				console.error("Failed to process Email Sending event", {
					eventId: message.body.payload.eventId,
					error,
				});
				message.retry();
			}
		}
	},
};

function shouldRemove(event) {
	if (event.type === "cf.email.sending.message.complained") {
		return true;
	}

	return (
		event.type === "cf.email.sending.message.bounced" &&
		event.payload.bounce?.type === "hard"
	);
}

async function removeRecipient(env, event) {
	await env.RECIPIENTS.delete(event.payload.recipient);
	console.log("Removed recipient record", {
		eventId: event.payload.eventId,
		reason: event.type,
	});
}

src/index.tsts

interface Env {
	RECIPIENTS: KVNamespace;
}

interface EmailSendingEvent {
	type:
		| "cf.email.sending.message.bounced"
		| "cf.email.sending.message.complained";
	payload: {
		eventId: string;
		recipient: string;
		bounce?: {
			type: "hard" | "soft";
		};
	};
}

export default {
	async queue(batch, env): Promise<void> {
		for (const message of batch.messages) {
			try {
				const event = message.body;

				if (shouldRemove(event)) {
					await removeRecipient(env, event);
				}

				message.ack();
			} catch (error) {
				console.error("Failed to process Email Sending event", {
					eventId: message.body.payload.eventId,
					error,
				});
				message.retry();
			}
		}
	},
} satisfies ExportedHandler<Env, EmailSendingEvent>;

function shouldRemove(event: EmailSendingEvent): boolean {
	if (event.type === "cf.email.sending.message.complained") {
		return true;
	}

	return (
		event.type === "cf.email.sending.message.bounced" &&
		event.payload.bounce?.type === "hard"
	);
}

async function removeRecipient(
	env: Env,
	event: EmailSendingEvent,
): Promise<void> {
	await env.RECIPIENTS.delete(event.payload.recipient);
	console.log("Removed recipient record", {
		eventId: event.payload.eventId,
		reason: event.type,
	});
}

删除不存在的 KV 键会成功。这使重复投递事件是安全的。

处理程序会确认每条成功的消息。失败操作在三次重试后会移至死信队列。

部署 Worker

部署 Worker 及其队列消费者配置：

npx wrangler deploy

yarn wrangler deploy

pnpm wrangler deploy

监控死信队列中的失败事件。修复根本错误后重新处理它们。

探索相关资源

事件订阅 — 查看事件 schema。
抑制列表 — 了解自动抑制。
Queues 重试 — 控制消息重试。
Workers KV 一致性 — 考虑传播延迟。