22.6 插件通信与协作

约 2045 字大约 7 分钟

概述

在复杂的插件系统中，插件之间的通信与协作是实现高级功能的关键。本章节将详细介绍 Claude Code 插件系统中的各种通信机制、协作模式和最佳实践。

插件通信机制

1. 事件总线 (Event Bus)

事件总线是插件间通信的核心机制，基于发布-订阅模式：

/**
 * 事件总线接口
 */
export interface IEventBus {
  /**
   * 发布事件
   * @param event 事件名称
   * @param data 事件数据
   */
  publish(event: string, data?: any): Promise<void>;

  /**
   * 订阅事件
   * @param event 事件名称
   * @param handler 事件处理函数
   */
  subscribe(event: string, handler: EventHandler): Subscription;

  /**
   * 取消订阅
   * @param subscription 订阅对象
   */
  unsubscribe(subscription: Subscription): void;

  /**
   * 订阅一次性事件
   * @param event 事件名称
   * @param handler 事件处理函数
   */
  once(event: string, handler: EventHandler): Subscription;
}

使用示例

// 插件A发布事件
context.eventBus.publish('user.created', { id: '123', name: 'John' });

// 插件B订阅事件
const subscription = context.eventBus.subscribe('user.created', (data) => {
  console.log('User created:', data);
});

// 取消订阅
subscription.unsubscribe();

2. 消息队列 (Message Queue)

消息队列用于异步通信和任务调度：

/**
 * 消息队列接口
 */
export interface IMessageQueue {
  /**
   * 发送消息
   * @param queue 队列名称
   * @param message 消息内容
   */
  send(queue: string, message: Message): Promise<MessageId>;

  /**
   * 接收消息
   * @param queue 队列名称
   * @param handler 消息处理函数
   */
  receive(queue: string, handler: MessageHandler): void;

  /**
   * 发送延迟消息
   * @param queue 队列名称
   * @param message 消息内容
   * @param delay 延迟时间（毫秒）
   */
  sendDelayed(queue: string, message: Message, delay: number): Promise<MessageId>;

  /**
   * 发送重试消息
   * @param queue 队列名称
   * @param message 消息内容
   * @param retryOptions 重试选项
   */
  sendWithRetry(queue: string, message: Message, retryOptions: RetryOptions): Promise<MessageId>;
}

使用示例

// 发送任务到队列
const messageId = await context.messageQueue.send('email.tasks', {
  to: 'user@example.com',
  subject: 'Welcome',
  body: 'Hello from Claude Code'
});

// 处理队列消息
context.messageQueue.receive('email.tasks', async (message) => {
  try {
    await sendEmail(message.data);
    return MessageStatus.SUCCESS;
  } catch (error) {
    return MessageStatus.FAILED;
  }
});

3. RPC 调用 (Remote Procedure Call)

RPC 允许插件直接调用其他插件的方法：

/**
 * RPC 客户端接口
 */
export interface IRPCClient {
  /**
   * 调用远程方法
   * @param service 服务名称
   * @param method 方法名称
   * @param parameters 方法参数
   */
  call(service: string, method: string, parameters?: any[]): Promise<any>;

  /**
   * 调用远程方法（带超时）
   * @param service 服务名称
   * @param method 方法名称
   * @param parameters 方法参数
   * @param timeout 超时时间（毫秒）
   */
  callWithTimeout(service: string, method: string, parameters?: any[], timeout?: number): Promise<any>;

  /**
   * 订阅服务
   * @param service 服务名称
   * @param callback 服务状态变化回调
   */
  subscribeService(service: string, callback: ServiceStatusCallback): void;
}

使用示例

// 插件A提供服务
context.rpcServer.register('calculator', {
  add: (a, b) => a + b,
  multiply: (a, b) => a * b
});

// 插件B调用服务
const result = await context.rpcClient.call('calculator', 'add', [2, 3]);
console.log('Result:', result); // 输出: 5

插件协作模式

1. 插件依赖管理

插件可以声明对其他插件的依赖：

// package.json
{
  "name": "my-plugin",
  "version": "1.0.0",
  "dependencies": {
    "auth-plugin": "^1.0.0",
    "storage-plugin": "^2.0.0"
  }
}

2. 服务发现与注册

插件可以注册服务供其他插件发现和使用：

// 注册服务
context.serviceRegistry.register('payment-service', {
  name: 'Stripe Payment',
  version: '1.0.0',
  endpoint: 'stripe-payment',
  metadata: { supportedCurrencies: ['USD', 'EUR', 'CNY'] }
});

// 发现服务
const services = await context.serviceRegistry.discover('payment-service');
const stripeService = services.find(s => s.metadata.supportedCurrencies.includes('USD'));

3. 插件链 (Plugin Chain)

插件可以形成链式处理流程：

// 插件A处理请求
context.eventBus.subscribe('request.received', async (data, next) => {
  data.userId = '123';
  await next(); // 传递给下一个插件
});

// 插件B处理请求
context.eventBus.subscribe('request.received', async (data, next) => {
  data.timestamp = Date.now();
  await next(); // 传递给下一个插件
});

// 执行插件链
const result = await context.pluginChain.execute('request.received', { url: '/api/users' });

数据共享与同步

1. 分布式缓存

插件可以使用分布式缓存共享数据：

// 设置缓存
await context.cache.set('user:123', { id: '123', name: 'John' }, { ttl: 3600 });

// 获取缓存
const user = await context.cache.get('user:123');

// 删除缓存
await context.cache.delete('user:123');

// 批量操作
await context.cache.multi()
  .set('user:123', user1)
  .set('user:456', user2)
  .expire('user:123', 3600)
  .exec();

2. 分布式锁

插件可以使用分布式锁确保数据一致性：

// 获取锁
const lock = await context.lock.acquire('resource:lock', { ttl: 30000 });

try {
  // 执行临界区操作
  await updateResource();
} finally {
  // 释放锁
  await lock.release();
}

3. 数据同步

插件可以使用数据同步机制保持数据一致性：

// 监听数据变化
context.dataSync.watch('users', async (change) => {
  if (change.type === 'create') {
    await sendWelcomeEmail(change.data);
  } else if (change.type === 'update') {
    await updateUserCache(change.data);
  }
});

// 同步数据
await context.dataSync.sync('users', {
  from: '2024-01-01',
  to: '2024-01-31'
});

事务处理

1. 分布式事务

插件可以参与分布式事务：

// 开始事务
const transaction = await context.transactionManager.begin();

try {
  // 执行多个操作
  await pluginA.createUser(transaction, userData);
  await pluginB.createOrder(transaction, orderData);
  await pluginC.sendNotification(transaction, notificationData);

  // 提交事务
  await transaction.commit();
} catch (error) {
  // 回滚事务
  await transaction.rollback();
  throw error;
}

2. 两阶段提交

插件可以实现两阶段提交协议：

// 第一阶段：准备
const participants = [pluginA, pluginB, pluginC];
const prepareResults = await Promise.all(
  participants.map(p => p.prepare(transaction))
);

// 检查准备结果
if (prepareResults.every(r => r.status === 'prepared')) {
  // 第二阶段：提交
  await Promise.all(participants.map(p => p.commit(transaction)));
} else {
  // 第二阶段：回滚
  await Promise.all(participants.map(p => p.rollback(transaction)));
}

错误处理与恢复

1. 错误传播

插件可以通过事件总线传播错误：

// 插件A抛出错误
try {
  await riskyOperation();
} catch (error) {
  context.eventBus.publish('plugin.error', {
    pluginId: 'plugin-a',
    error: error.message,
    stack: error.stack
  });
  throw error;
}

// 插件B监听错误
context.eventBus.subscribe('plugin.error', (errorData) => {
  context.logger.error('Plugin error:', errorData);
  // 执行错误恢复逻辑
});

2. 熔断机制

插件可以使用熔断机制防止级联失败：

// 创建熔断器
const circuitBreaker = context.circuitBreakerFactory.create('external-service', {
  failureThreshold: 50,
  successThreshold: 10,
  timeout: 5000
});

// 使用熔断器
try {
  const result = await circuitBreaker.execute(() => callExternalService());
  return result;
} catch (error) {
  if (error.type === 'CircuitOpenError') {
    // 熔断器打开，返回降级结果
    return fallbackResult();
  }
  throw error;
}

3. 降级策略

插件可以实现降级策略：

// 降级策略示例
async function getUsers() {
  try {
    return await database.query('SELECT * FROM users');
  } catch (error) {
    // 数据库失败，返回缓存数据
    context.logger.warn('Database failed, falling back to cache');
    return cache.get('users:all');
  }
}

性能优化

1. 批量处理

插件可以批量处理请求提高性能：

// 批量处理事件
const batchProcessor = context.batchProcessor.create('user-events', {
  batchSize: 100,
  batchTimeout: 1000
});

// 发送事件
batchProcessor.add({ type: 'created', userId: '123' });
batchProcessor.add({ type: 'updated', userId: '456' });

// 处理批量
batchProcessor.on('batch', async (events) => {
  await database.batchInsert('user_events', events);
});

2. 异步处理

插件可以使用异步处理提高响应性：

// 异步处理请求
app.post('/api/users', async (req, res) => {
  // 立即返回响应
  res.status(202).json({ status: 'accepted' });

  // 异步处理任务
  setTimeout(async () => {
    await createUser(req.body);
    await sendWelcomeEmail(req.body.email);
  }, 0);
});

3. 缓存策略

插件可以使用多级缓存提高性能：

// 多级缓存
async function getUser(userId) {
  // 先查本地缓存
  let user = localCache.get(`user:${userId}`);
  if (user) return user;

  // 再查分布式缓存
  user = await distributedCache.get(`user:${userId}`);
  if (user) {
    localCache.set(`user:${userId}`, user, { ttl: 60 });
    return user;
  }

  // 最后查数据库
  user = await database.query('SELECT * FROM users WHERE id = ?', [userId]);
  distributedCache.set(`user:${userId}`, user, { ttl: 3600 });
  localCache.set(`user:${userId}`, user, { ttl: 60 });
  return user;
}

安全考虑

1. 权限控制

插件可以实现细粒度的权限控制：

// 权限检查
async function deleteUser(userId, requester) {
  const permission = await context.permissionService.check(
    requester.id,
    'user:delete',
    { userId }
  );

  if (!permission.granted) {
    throw new Error('Permission denied');
  }

  await database.query('DELETE FROM users WHERE id = ?', [userId]);
}

2. 数据加密

插件可以加密敏感数据：

// 加密数据
const encryptedData = await context.encryptionService.encrypt(
  sensitiveData,
  'AES-256-GCM',
  { keyId: 'data-encryption-key' }
);

// 解密数据
const decryptedData = await context.encryptionService.decrypt(
  encryptedData,
  'AES-256-GCM',
  { keyId: 'data-encryption-key' }
);

3. 审计日志

插件可以记录审计日志：

// 记录审计日志
context.auditLogger.log({
  action: 'user.created',
  actor: 'system',
  target: userId,
  metadata: { ipAddress: '192.168.1.1', userAgent: 'Claude Code' },
  timestamp: new Date()
});

最佳实践

1. 通信协议设计

使用标准化的事件名称和消息格式
定义清晰的版本控制策略
提供详细的文档和示例

2. 错误处理

始终处理异步操作的错误
提供有意义的错误信息
实现适当的重试和降级策略

3. 性能优化

批量处理减少系统调用
使用缓存减少重复计算
异步处理提高响应性

4. 安全性

验证所有外部输入
实施最小权限原则
加密敏感数据

5. 可观测性

记录关键操作和事件
监控插件性能和健康状况
提供详细的错误信息

总结

插件通信与协作是构建复杂插件系统的关键。通过选择合适的通信机制、遵循协作模式和最佳实践，可以构建出高效、可靠和安全的插件系统。

下一章将介绍插件测试与调试技术，帮助开发者确保插件质量和稳定性。