架构之异步

异步架构是构建高性能、高并发系统的关键技术。通过合理选择异步编程模型、设计异步API、使用消息队列和事件驱动架构，可以显著提升系统的吞吐量和响应能力。关键要点选择合适的异步模型：根据场景选择Promise、async/await、响应式编程等设计清晰的异步API：明确状态、超时、错误处理使用消息队列解耦：通过生产者-消费者、发布-订阅模式实现松耦合关注错误处理：重试、断路器、降级等策略监控与可观测

dijk

766人浏览 · 2026-01-27 16:25:54

dijk · 2026-01-27 16:25:54 发布

架构之异步

核心概念

什么是异步架构

异步架构是一种系统设计方法，其中任务的执行不依赖于调用线程的同步等待。相反，系统可以在后台处理任务的同时继续处理其他请求。

同步 vs 异步

特性	同步	异步
执行方式	阻塞等待结果	非阻塞，回调或Promise
资源利用率	低（线程等待）	高（线程可复用）
响应时间	线性累积	可并发处理
复杂度	简单直观	需要状态管理
适用场景	快速操作	IO密集型、耗时操作

异步架构的价值

提升吞吐量：通过非阻塞IO，单个线程可以处理大量并发连接
降低延迟：避免线程阻塞，提高响应速度
资源高效：减少线程上下文切换开销
弹性扩展：天然适合分布式系统
解耦组件：通过消息传递实现松耦合

异步编程模型

回调模式

最基础的异步编程方式，通过回调函数处理异步结果。

// JavaScript 示例
function fetchData(url, callback) {
  setTimeout(() => {
    const data = { id: 1, name: 'Async Data' };
    callback(null, data);
  }, 100);
}

fetchData('https://api.example.com', (error, data) => {
  if (error) {
    console.error('Error:', error);
  } else {
    console.log('Data:', data);
  }
});

优点：简单直接，所有语言都支持

缺点：回调地狱，错误处理困难

Promise/Future

提供链式调用和更好的错误处理机制。

// JavaScript Promise
function fetchData(url) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve({ id: 1, name: 'Async Data' });
    }, 100);
  });
}

fetchData('https://api.example.com')
  .then(data => console.log('Data:', data))
  .catch(error => console.error('Error:', error));

// Java CompletableFuture
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 模拟耗时操作
    try { Thread.sleep(100); } catch (InterruptedException e) {}
    return "Async Result";
});

future.thenAccept(result -> System.out.println("Result: " + result))
      .exceptionally(error -> {
          System.err.println("Error: " + error);
          return null;
      });

Async/Await

语法糖，让异步代码看起来像同步代码。

// JavaScript async/await
async function main() {
  try {
    const data = await fetchData('https://api.example.com');
    console.log('Data:', data);
  } catch (error) {
    console.error('Error:', error);
  }
}

# Python async/await
import asyncio

async def fetch_data(url):
    await asyncio.sleep(0.1)  # 模拟IO操作
    return {'id': 1, 'name': 'Async Data'}

async def main():
    try:
        data = await fetch_data('https://api.example.com')
        print(f'Data: {data}')
    except Exception as error:
        print(f'Error: {error}')

asyncio.run(main())

响应式编程

基于数据流和变化传播的编程范式。

// Project Reactor 示例
Flux.just("item1", "item2", "item3")
    .map(String::toUpperCase)
    .filter(s -> s.startsWith("ITEM"))
    .subscribe(
        item -> System.out.println("Received: " + item),
        error -> System.err.println("Error: " + error),
        () -> System.out.println("Completed")
    );

异步架构模式

生产者-消费者模式

最经典的异步模式，通过队列解耦生产者和消费者。

┌─────────┐      ┌─────────┐      ┌─────────┐
│Producer │ ───> │  Queue  │ ───> │Consumer │
└─────────┘      └─────────┘      └─────────┘

应用场景：

日志收集
数据处理管道
任务调度
事件通知

发布-订阅模式

一对多的消息传递模式，多个消费者可以订阅同一个主题。

         ┌─────────┐
         │Publisher│
         └────┬────┘
              │
         ┌────▼────┐
         │  Topic  │
         └────┬────┘
        ┌─────┼─────┐
        ▼     ▼     ▼
  ┌───────┐ ┌───────┐ ┌───────┐
  │Sub1   │ │Sub2   │ │Sub3   │
  └───────┘ └───────┘ └───────┘

应用场景：

实时通知
数据广播
事件溯源
微服务通信

CQRS（命令查询职责分离）

将读操作和写操作分离，使用不同的模型和存储。

┌─────────────┐      ┌─────────────┐
│   Command   │ ───> │ Write Model  │
│   Service   │      │ (Event Store)│
└─────────────┘      └──────┬──────┘
                            │ Events
                            ▼
                     ┌─────────────┐
                     │  Read Model │
                     │  (Cache)    │
                     └──────┬──────┘
                            │
┌─────────────┐      ┌──────▼──────┐
│   Query     │ ───> │ Read Service│
│   Service   │      │             │
└─────────────┘      └─────────────┘

事件溯源

通过存储事件流来重建状态，而非直接存储当前状态。

Event Store:
┌─────────────────────────────────┐
│ Event1: OrderCreated            │
│ Event2: ItemAdded               │
│ Event3: PaymentCompleted        │
│ Event4: OrderShipped            │
└─────────────────────────────────┘
         │
         │ Replay Events
         ▼
┌─────────────────┐
│ Current State   │
│ Order: Shipped  │
└─────────────────┘

消息队列与异步通信

主流消息队列对比

特性	RabbitMQ	Kafka	Redis Stream	RocketMQ
消息模型	队列/主题	日志流	流	队列/主题
吞吐量	中	极高	高	高
延迟	低	低	极低	低
持久化	支持	支持	支持	支持
回溯消费	困难	支持	支持	支持
消息顺序	支持	分区内有序	支持	支持
适用场景	业务解耦	日志/流处理	轻量级队列	电商/金融

RabbitMQ 架构

┌──────────┐      ┌──────────┐      ┌──────────┐
│Producer  │ ───> │Exchange  │ ───> │Queue     │
└──────────┘      └────┬─────┘      └────┬─────┘
                       │                 │
                       │ Routing Key     │
                       ▼                 ▼
                 ┌──────────┐      ┌──────────┐
                 │Queue     │      │Consumer  │
                 └────┬─────┘      └──────────┘
                      │
                      ▼
                 ┌──────────┐
                 │Consumer  │
                 └──────────┘

核心概念：

Exchange：消息路由器，支持多种路由策略
Queue：消息存储容器
Binding：Exchange与Queue的绑定关系
Routing Key：路由规则

Exchange 类型：

Direct：精确匹配路由键
Topic：通配符匹配路由键
Fanout：广播到所有绑定的队列
Headers：基于消息头匹配

Kafka 架构

┌─────────────────────────────────────────────────┐
│                  Kafka Cluster                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │Broker 1  │  │Broker 2  │  │Broker 3  │       │
│  │Topic A   │  │Topic A   │  │Topic A   │       │
│  │Partition0│  │Partition1│  │Partition2│       │
│  └──────────┘  └──────────┘  └──────────┘       │
└─────────────────────────────────────────────────┘
        ▲                                    ▲
        │                                    │
┌───────┴────────┐                  ┌────────┴────────┐
│   Producer     │                  │    Consumer     │
│                │                  │   (Consumer     │
│                │                  │    Group)       │
└────────────────┘                  └─────────────────┘

核心概念：

Topic：消息分类
Partition：Topic的分区，实现并行消费
Offset：消息在分区中的位置
Consumer Group：消费者组，实现负载均衡
Replica：分区副本，实现高可用

事件驱动架构

架构组成

┌─────────────────────────────────────────────────────┐
│                   Event Bus                          │
│              (Kafka/RabbitMQ/NATS)                   │
└─────────────────────────────────────────────────────┘
         ▲              ▲              ▲
         │              │              │
    ┌────┴────┐   ┌─────┴─────┐   ┌────┴────┐
    │Service A│   │Service B  │   │Service C│
    │(Producer│   │(Producer  │   │(Consumer│
    │+Consumer)│  │+Consumer) │   │ Only)   │
    └─────────┘   └───────────┘   └─────────┘

事件类型

领域事件：业务领域内发生的重要事件
- OrderCreated
- PaymentCompleted
- UserRegistered
集成事件：跨服务边界的业务事件
- OrderPaid (触发库存扣减)
- ShipmentDelivered (触发通知)
系统事件：基础设施相关事件
- ServiceHealthCheck
- MetricCollected

事件设计原则

// 标准事件格式
{
  "eventId": "550e8400-e29b-41d4-a716-446655440000",
  "eventType": "OrderCreated",
  "eventVersion": "1.0",
  "occurredAt": "2024-01-27T10:30:00Z",
  "aggregateId": "order-12345",
  "aggregateType": "Order",
  "data": {
    "orderId": "order-12345",
    "customerId": "customer-67890",
    "items": [
      {
        "productId": "prod-001",
        "quantity": 2,
        "price": 99.99
      }
    ],
    "totalAmount": 199.98
  },
  "metadata": {
    "correlationId": "req-abc-123",
    "causationId": "req-abc-123",
    "source": "order-service"
  }
}

最终一致性处理

在事件驱动架构中，数据最终通过事件传播达到一致状态。

处理策略：

幂等消费：使用唯一ID去重
补偿事务：通过反向事件回滚
Saga模式：长事务分解为多个本地事务
重试机制：指数退避重试

异步任务调度

延迟任务

任务在指定时间后执行。

// Java ScheduledExecutorService
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

// 延迟1秒执行
scheduler.schedule(() -> {
    System.out.println("Delayed task executed");
}, 1, TimeUnit.SECONDS);

// 固定延迟执行
scheduler.scheduleWithFixedDelay(() -> {
    System.out.println("Fixed delay task");
}, 0, 5, TimeUnit.SECONDS);

// 固定频率执行
scheduler.scheduleAtFixedRate(() -> {
    System.out.println("Fixed rate task");
}, 0, 5, TimeUnit.SECONDS);

定时任务

按固定时间表执行的任务。

Cron表达式：

* * * * * *
│ │ │ │ │ │
│ │ │ │ │ └─ 星期几 (0-7, 0和7都表示周日)
│ │ │ │ └─── 月份 (1-12)
│ │ │ └───── 日期 (1-31)
│ │ └─────── 小时 (0-23)
│ └───────── 分钟 (0-59)
└─────────── 秒 (0-59)

常用Cron示例：

0 0 12 * * ? - 每天中午12点
0 0/5 * * * ? - 每5分钟
0 0 2 * * ? - 每天凌晨2点
0 0 0 1 * ? - 每月1号零点

分布式任务调度

在集群环境中协调任务执行，避免重复执行。

实现方式：

数据库锁

-- 获取锁
SELECT * FROM task_lock WHERE task_name = 'daily-report' FOR UPDATE;

-- 释放锁
COMMIT;

Redis分布式锁

import redis
import time

def acquire_lock(redis_client, lock_name, expire_time=10):
    """获取分布式锁"""
    identifier = str(time.time())
    if redis_client.setnx(lock_name, identifier):
        redis_client.expire(lock_name, expire_time)
        return identifier
    return None

def release_lock(redis_client, lock_name, identifier):
    """释放分布式锁"""
    pipe = redis_client.pipeline(True)
    while True:
        try:
            pipe.watch(lock_name)
            if pipe.get(lock_name) == identifier:
                pipe.multi()
                pipe.delete(lock_name)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.WatchError:
            pass
    return False

ZooKeeper临时节点

// ZooKeeper分布式锁
public class DistributedLock {
    private ZooKeeper zk;
    private String lockPath;
    
    public boolean tryLock() throws Exception {
        String path = zk.create(lockPath + "/lock-", 
                               new byte[0],
                               ZooDefs.Ids.OPEN_ACL_UNSAFE,
                               CreateMode.EPHEMERAL_SEQUENTIAL);
        // 检查是否是最小的节点
        List<String> children = zk.getChildren(lockPath, false);
        Collections.sort(children);
        if (path.equals(lockPath + "/" + children.get(0))) {
            return true;
        }
        // 否则等待前一个节点释放
        return false;
    }
}

异步API设计

异步HTTP API

1. 异步请求-响应模式

客户端发起请求后立即返回，通过回调或轮询获取结果。

// 提交异步任务
POST /api/v1/jobs HTTP/1.1
Content-Type: application/json

{
  "type": "report-generation",
  "parameters": {
    "startDate": "2024-01-01",
    "endDate": "2024-01-31"
  }
}

// 响应
HTTP/1.1 202 Accepted
Location: /api/v1/jobs/job-12345

{
  "jobId": "job-12345",
  "status": "pending",
  "createdAt": "2024-01-27T10:00:00Z"
}

// 轮询任务状态
GET /api/v1/jobs/job-12345 HTTP/1.1

{
  "jobId": "job-12345",
  "status": "completed",
  "progress": 100,
  "result": {
    "reportUrl": "https://cdn.example.com/reports/report-12345.pdf"
  },
  "completedAt": "2024-01-27T10:05:00Z"
}

2. Webhook回调模式

任务完成后，服务端主动通知客户端。

// 提交任务时指定回调URL
POST /api/v1/jobs HTTP/1.1
Content-Type: application/json

{
  "type": "video-processing",
  "parameters": {
    "videoUrl": "https://storage.example.com/video.mp4"
  },
  "callbackUrl": "https://client.example.com/webhooks/job-completed",
  "callbackSecret": "webhook-secret-key"
}

// 任务完成后回调
POST https://client.example.com/webhooks/job-completed HTTP/1.1
Content-Type: application/json
X-Webhook-Signature: sha256=abc123...

{
  "jobId": "job-12345",
  "status": "completed",
  "result": {
    "processedUrl": "https://storage.example.com/processed-video.mp4",
    "thumbnailUrl": "https://storage.example.com/thumbnail.jpg"
  },
  "completedAt": "2024-01-27T10:05:00Z"
}

3. Server-Sent Events (SSE)

服务器向客户端推送事件流。

// 客户端订阅
GET /api/v1/events/stream HTTP/1.1
Accept: text/event-stream

// 服务端响应
HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache

data: {"type":"job_started","jobId":"job-12345"}

data: {"type":"job_progress","jobId":"job-12345","progress":25}

data: {"type":"job_progress","jobId":"job-12345","progress":50}

data: {"type":"job_completed","jobId":"job-12345","result":{...}}

4. WebSocket双向通信

全双工通信，支持实时交互。

// 客户端连接
const ws = new WebSocket('wss://api.example.com/ws/jobs');

ws.onopen = () => {
  // 提交任务
  ws.send(JSON.stringify({
    type: 'submit_job',
    job: {
      type: 'data-analysis',
      parameters: { /* ... */ }
    }
  }));
};

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  switch (message.type) {
    case 'job_started':
      console.log('Job started:', message.jobId);
      break;
    case 'job_progress':
      console.log('Progress:', message.progress);
      break;
    case 'job_completed':
      console.log('Result:', message.result);
      break;
  }
};

异步API最佳实践

明确的HTTP状态码
- 202 Accepted - 任务已接受，正在处理
- 200 OK - 任务已完成，返回结果
- 404 Not Found - 任务不存在
- 409 Conflict - 任务状态不允许此操作
合理的超时设置
- 请求超时：30秒
- 轮询间隔：指数退避（1s, 2s, 4s, 8s…）
- 任务超时：根据业务设置（如1小时）
幂等性设计
- 使用唯一ID标识任务
- 重复提交返回相同结果

错误处理

{
  "error": {
    "code": "TASK_FAILED",
    "message": "Task processing failed",
    "details": {
      "reason": "Invalid input parameters",
      "retryable": false
    }
  }
}

最佳实践

1. 选择合适的异步模型

场景	推荐模型
简单异步操作	Promise/Future
复杂异步流程	Async/Await
大规模数据流	响应式编程
跨服务通信	消息队列
实时推送	WebSocket/SSE

2. 错误处理策略

// 重试模式
async function fetchWithRetry(url, maxRetries = 3, delay = 1000) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      const response = await fetch(url);
      if (response.ok) return response;
      throw new Error(`HTTP ${response.status}`);
    } catch (error) {
      if (i === maxRetries - 1) throw error;
      await new Promise(resolve => 
        setTimeout(resolve, delay * Math.pow(2, i))
      );
    }
  }
}

// 断路器模式
class CircuitBreaker {
  constructor(threshold = 5, timeout = 60000) {
    this.failureCount = 0;
    this.threshold = threshold;
    this.timeout = timeout;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
    this.nextAttempt = 0;
  }

  async execute(fn) {
    if (this.state === 'OPEN') {
      if (Date.now() > this.nextAttempt) {
        this.state = 'HALF_OPEN';
      } else {
        throw new Error('Circuit breaker is OPEN');
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  onSuccess() {
    this.failureCount = 0;
    this.state = 'CLOSED';
  }

  onFailure() {
    this.failureCount++;
    if (this.failureCount >= this.threshold) {
      this.state = 'OPEN';
      this.nextAttempt = Date.now() + this.timeout;
    }
  }
}

3. 背压处理

在响应式编程中，处理生产者速度快于消费者速度的情况。

// Reactor 背压策略
Flux.range(1, 1000)
    .onBackpressureBuffer(100)          // 缓冲100个元素
    .onBackpressureDrop()               // 丢弃无法处理的元素
    .onBackpressureLatest()             // 只保留最新元素
    .subscribe(i -> process(i));

4. 资源管理

// 使用连接池
const { Pool } = require('pg');
const pool = new Pool({
  max: 20,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
});

// 优雅关闭
async function gracefulShutdown() {
  await pool.end();
  await messageQueue.close();
  await server.close();
}

process.on('SIGTERM', gracefulShutdown);
process.on('SIGINT', gracefulShutdown);

5. 监控与可观测性

// 添加指标
const metrics = {
  asyncOperations: new Counter(),
  asyncErrors: new Counter(),
  asyncDuration: new Histogram(),
};

async function trackAsyncOperation(fn) {
  const timer = metrics.asyncDuration.startTimer();
  try {
    const result = await fn();
    metrics.asyncOperations.inc();
    return result;
  } catch (error) {
    metrics.asyncErrors.inc();
    throw error;
  } finally {
    timer();
  }
}

常见陷阱与解决方案

1. 回调地狱

问题：多层嵌套回调导致代码难以维护

解决方案：使用Promise/async-await

// 回调地狱（不推荐）
fetchData(url1, (err1, data1) => {
  fetchData(url2 + data1.id, (err2, data2) => {
    fetchData(url3 + data2.id, (err3, data3) => {
      // ...
    });
  });
});

// 使用async-await（推荐）
async function fetchAll() {
  const data1 = await fetchData(url1);
  const data2 = await fetchData(url2 + data1.id);
  const data3 = await fetchData(url3 + data2.id);
  return data3;
}

2. 异常丢失

问题：异步代码中的异常可能被忽略

解决方案：确保所有Promise都被处理

// 错误示例
async function badExample() {
  const promise = fetchData(url);
  // 如果这里抛出异常，promise的异常会被忽略
  throw new Error('Something went wrong');
}

// 正确示例
async function goodExample() {
  const promise = fetchData(url);
  try {
    throw new Error('Something went wrong');
  } finally {
    await promise.catch(error => console.error(error));
  }
}

3. 竞态条件

问题：多个异步操作并发执行导致状态不一致

解决方案：使用锁或串行化

// 竞态条件
let counter = 0;
async function increment() {
  const current = counter;
  await delay(10);
  counter = current + 1;
}

// 解决方案：使用锁
class Mutex {
  constructor() {
    this.queue = [];
    this.locked = false;
  }

  async acquire() {
    while (this.locked) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    this.locked = true;
  }

  release() {
    if (this.queue.length > 0) {
      const next = this.queue.shift();
      next();
    } else {
      this.locked = false;
    }
  }
}

const mutex = new Mutex();
let counter = 0;

async function safeIncrement() {
  await mutex.acquire();
  try {
    const current = counter;
    await delay(10);
    counter = current + 1;
  } finally {
    mutex.release();
  }
}

4. 内存泄漏

问题：未正确清理异步资源导致内存持续增长

解决方案：正确取消和清理

// 错误示例
function startPolling() {
  setInterval(() => {
    fetchData(url).then(data => console.log(data));
  }, 1000);
  // interval无法被清理
}

// 正确示例
function startPolling() {
  const intervalId = setInterval(() => {
    fetchData(url).then(data => console.log(data));
  }, 1000);
  
  return () => clearInterval(intervalId);
}

// 使用
const stopPolling = startPolling();
// 需要时停止
stopPolling();

5. 过度异步

问题：简单的操作也使用异步，增加复杂度

解决方案：合理评估是否需要异步

// 不必要的异步
async function add(a, b) {
  return a + b;
}

// 同步即可
function add(a, b) {
  return a + b;
}

性能优化

1. 批量处理

将多个小任务合并为一个大任务处理。

// 单个处理（效率低）
for (const item of items) {
  await processItem(item);
}

// 批量处理（高效）
async function batchProcess(items, batchSize = 100) {
  for (let i = 0; i < items.length; i += batchSize) {
    const batch = items.slice(i, i + batchSize);
    await Promise.all(batch.map(item => processItem(item)));
  }
}

2. 并发控制

限制同时执行的异步操作数量。

async function concurrentLimit(tasks, limit = 10) {
  const results = [];
  const executing = [];

  for (const task of tasks) {
    const promise = task().then(result => {
      executing.splice(executing.indexOf(promise), 1);
      return result;
    });

    executing.push(promise);

    if (executing.length >= limit) {
      await Promise.race(executing);
    }
  }

  return Promise.all(executing);
}

3. 缓存策略

缓存异步操作结果，避免重复计算。

const cache = new Map();

async function cachedFetch(url) {
  if (cache.has(url)) {
    return cache.get(url);
  }

  const promise = fetch(url).then(response => response.json());
  cache.set(url, promise);
  
  try {
    return await promise;
  } catch (error) {
    cache.delete(url);
    throw error;
  }
}

4. 连接复用

复用数据库、HTTP等连接，减少建立连接的开销。

// 数据库连接池
const { Pool } = require('pg');
const pool = new Pool({
  max: 20,
  min: 5,
  idle: 10000,
});

// HTTP连接复用（keep-alive）
const https = require('https');
const agent = new https.Agent({
  keepAlive: true,
  maxSockets: 50,
  maxFreeSockets: 10,
});

fetch(url, { agent });

5. 异步IO优化

使用零拷贝、内存映射等技术优化IO性能。

// Java NIO 零拷贝文件传输
FileChannel sourceChannel = FileChannel.open(sourcePath);
FileChannel destChannel = FileChannel.open(destPath, 
    StandardOpenOption.CREATE, StandardOpenOption.WRITE);

// 直接传输，无需经过用户空间
sourceChannel.transferTo(0, sourceChannel.size(), destChannel);

总结

异步架构是构建高性能、高并发系统的关键技术。通过合理选择异步编程模型、设计异步API、使用消息队列和事件驱动架构，可以显著提升系统的吞吐量和响应能力。

关键要点

选择合适的异步模型：根据场景选择Promise、async/await、响应式编程等
设计清晰的异步API：明确状态、超时、错误处理
使用消息队列解耦：通过生产者-消费者、发布-订阅模式实现松耦合
关注错误处理：重试、断路器、降级等策略
监控与可观测性：跟踪异步操作的性能和状态
避免常见陷阱：回调地狱、竞态条件、内存泄漏等
持续优化性能：批量处理、并发控制、连接复用等

适用场景

高并发Web服务：Node.js、Netty等异步框架
微服务架构：通过消息队列实现服务间异步通信
实时系统：WebSocket、SSE等实时推送
数据处理管道：ETL、流处理等异步数据处理
事件溯源系统：通过事件流重建状态

异步架构不是万能的，需要根据具体业务场景和技术栈选择合适的方案。在追求性能的同时，也要考虑代码的可维护性和系统的可靠性。

2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

大模型工程师转型指南：别被学历吓退，薪资翻倍不是梦！非常详细收藏我这一篇就够了

2048 AI社区

AI时代：当机器成为人类的一面镜子

但这种便利的背面，是个人数据如同透明橱窗里的展品，是选择看似自由实则被算法精心引导的微妙控制。当AI的镜像越来越清晰地映照出人类的智慧，那镜中最耀眼的光芒，不应是算法的精妙，而是人性中无法被编码的部分：同理心、道德勇气、无目的的 curiosity，以及在不确定性面前依然选择善良的意志。人类第一次面对能“理解”我们却无意识的交流对象，这种不对称关系迫使我们重新审视：情感的本质上是一种生物化学反应，

2048 AI社区

DeepSeek-R1一周年，DeepSeek-R2要来了？

2048 AI社区

所有评论(0)

查看更多评论

dijk

@IOIO_

已为社区贡献72条内容

架构之异步

dijk

架构之异步

目录

核心概念

什么是异步架构

同步 vs 异步

异步架构的价值

异步编程模型

回调模式

Promise/Future

Async/Await

响应式编程

异步架构模式

生产者-消费者模式

发布-订阅模式

CQRS（命令查询职责分离）

事件溯源

消息队列与异步通信

主流消息队列对比

RabbitMQ 架构

Kafka 架构

事件驱动架构

架构组成

事件类型

事件设计原则

最终一致性处理

异步任务调度

延迟任务

定时任务

分布式任务调度

异步API设计

异步HTTP API

1. 异步请求-响应模式

2. Webhook回调模式

3. Server-Sent Events (SSE)

4. WebSocket双向通信

异步API最佳实践

最佳实践

1. 选择合适的异步模型

2. 错误处理策略

3. 背压处理

4. 资源管理

5. 监控与可观测性

常见陷阱与解决方案

1. 回调地狱

2. 异常丢失

3. 竞态条件

4. 内存泄漏

5. 过度异步

性能优化

1. 批量处理

2. 并发控制

3. 缓存策略

4. 连接复用

5. 异步IO优化

总结

关键要点

适用场景

所有评论(0)

dijk