使用 Apollo Server 开发 GraphQL 订阅：实现实时数据的魔法

引言

大家好，欢迎来到今天的讲座！今天我们要一起探讨如何使用 Apollo Server 来开发 GraphQL 订阅，实现真正的实时数据推送。如果你对 GraphQL 有一定的了解，但还没有接触过订阅功能，那么你来对地方了！我们将从基础开始，一步步带你走进这个神奇的世界，让你不仅能理解订阅的工作原理，还能亲手写出一个完整的实时应用。

在接下来的时间里，我们会通过轻松诙谐的语言、通俗易懂的解释，以及大量的代码示例，帮助你掌握这个强大的工具。准备好了吗？让我们开始吧！✨

什么是 GraphQL 订阅？

1. 订阅的基本概念

首先，我们来回顾一下 GraphQL 的核心概念。GraphQL 是一种用于 API 查询的语言和运行时，它允许客户端精确地请求所需的数据，而不需要像 REST 那样依赖固定的端点。这使得开发者可以更灵活地获取数据，减少不必要的网络请求和数据传输。

然而，传统的 GraphQL 查询和变更（mutation）都是 单向的：客户端发起请求，服务器返回响应。这种模式适用于大多数场景，但对于需要 实时更新 的应用场景，比如聊天应用、股票行情、在线游戏等，就显得不够灵活了。这时，我们就需要用到 GraphQL 订阅。

GraphQL 订阅 是一种特殊的查询类型，它允许客户端 持续监听 服务器上的某些事件，并在这些事件发生时立即收到通知。换句话说，订阅可以让服务器主动将最新的数据推送给客户端，而不是等待客户端再次发起请求。这就实现了真正的 实时数据流。

2. 订阅的工作原理

为了理解订阅的工作原理，我们可以将其类比为广播电台和收音机之间的关系。假设你正在听一个广播电台，电台会不断播放最新的新闻、音乐等内容，而你只需要打开收音机，就可以随时接收到这些信息。同样，客户端通过订阅某个事件，就像是打开了一个“收音机”，服务器则像广播电台一样，一旦有新的数据产生，就会立即推送给所有订阅了该事件的客户端。

在技术层面上，订阅通常通过 WebSocket 实现。WebSocket 是一种双向通信协议，允许服务器和客户端之间保持持久连接，从而实现低延迟的数据传输。与 HTTP 不同，WebSocket 不需要每次请求都重新建立连接，因此非常适合用于实时数据推送。

3. 订阅的优势

• 实时性：订阅可以让客户端在数据发生变化时立即收到通知，而不需要轮询或手动刷新页面。
• 高效性：由于 WebSocket 连接是持久的，服务器可以在事件发生时立即推送数据，减少了不必要的网络请求。
• 灵活性：订阅可以根据不同的条件进行过滤，客户端可以选择只接收自己感兴趣的数据。
• 简化开发：通过 Apollo Server 和 GraphQL，你可以非常方便地实现订阅功能，而不需要编写复杂的 WebSocket 逻辑。

准备工作

在我们开始编写代码之前，确保你已经安装了以下工具：

• Node.js：版本 14 或更高。
• npm 或 yarn：用于管理项目依赖。
• Apollo Server：我们将使用 Apollo Server 来搭建 GraphQL 服务。
• GraphQL：确保你已经对 GraphQL 有基本的了解，特别是查询和变更的操作。

如果你还没有安装这些工具，可以通过以下命令快速安装：

# 安装 Node.js 和 npm
curl -fsSL https://deb.nodesource.com/setup_16.x | sudo -E bash -
sudo apt-get install -y nodejs

# 安装 yarn（可选）
npm install --global yarn

接下来，创建一个新的项目目录，并初始化项目：

mkdir graphql-subscriptions
cd graphql-subscriptions
npm init -y

安装所需的依赖：

npm install apollo-server graphql ws

apollo-server 是我们用来搭建 GraphQL 服务的核心库，graphql 是 GraphQL 规范的 JavaScript 实现，ws 是一个轻量级的 WebSocket 库，用于处理订阅的实时通信。

创建一个简单的 GraphQL 服务

在开始编写订阅功能之前，我们先创建一个简单的 GraphQL 服务，以便后续添加订阅功能。我们将实现一个基本的查询和变更操作，模拟一个简单的任务管理系统。

1. 定义 GraphQL 类型

首先，我们需要定义 GraphQL 的类型系统。在 schema.js 文件中，定义任务（Task）的类型以及查询和变更的接口：

const { gql } = require('apollo-server');

const typeDefs = gql`
  type Task {
    id: ID!
    title: String!
    completed: Boolean!
  }

  type Query {
    tasks: [Task!]!
  }

  type Mutation {
    addTask(title: String!): Task!
    completeTask(id: ID!): Task!
  }
`;

module.exports = typeDefs;

这里我们定义了一个 Task 类型，包含 id、title 和 completed 三个字段。Query 类型允许客户端查询所有任务，而 Mutation 类型提供了添加任务和标记任务为完成的功能。

2. 实现解析器

接下来，我们需要实现解析器（resolvers），即告诉 Apollo Server 如何处理查询和变更请求。在 resolvers.js 文件中，编写如下代码：

const tasks = [];

const resolvers = {
  Query: {
    tasks: () => tasks,
  },
  Mutation: {
    addTask: (_, { title }) => {
      const newTask = {
        id: Math.random().toString(36).substr(2, 9),
        title,
        completed: false,
      };
      tasks.push(newTask);
      return newTask;
    },
    completeTask: (_, { id }) => {
      const task = tasks.find(t => t.id === id);
      if (!task) throw new Error('Task not found');
      task.completed = true;
      return task;
    },
  },
};

module.exports = resolvers;

在这个例子中，我们使用了一个简单的数组 tasks 来存储任务。addTask 解析器会在每次调用时生成一个新的任务并将其添加到数组中，completeTask 解析器则会查找指定的任务并将其标记为完成。

3. 启动 Apollo Server

现在我们已经有了类型定义和解析器，接下来可以启动 Apollo Server 了。在 index.js 文件中，编写如下代码：

const { ApolloServer } = require('apollo-server');
const typeDefs = require('./schema');
const resolvers = require('./resolvers');

const server = new ApolloServer({
  typeDefs,
  resolvers,
});

server.listen().then(({ url }) => {
  console.log(`🚀  Server ready at ${url}`);
});

这段代码创建了一个 Apollo Server 实例，并将我们定义的类型和解析器传递给它。最后，服务器会在控制台输出一个 URL，表示 API 已经启动并准备好接收请求。

4. 测试查询和变更

启动服务器后，打开浏览器并访问 http://localhost:4000，你会看到 Apollo Server 提供的 GraphQL Playground 界面。在这里，你可以测试我们刚刚实现的查询和变更操作。

例如，添加一个新任务：

mutation {
  addTask(title: "Learn GraphQL Subscriptions") {
    id
    title
    completed
  }
}

查询所有任务：

query {
  tasks {
    id
    title
    completed
  }
}

如果一切正常，你应该能够成功添加任务并查询到它们。恭喜你，你已经完成了一个简单的 GraphQL 服务！🎉

添加订阅功能

现在我们有了一个基本的 GraphQL 服务，接下来要做的就是为其添加订阅功能。我们将实现一个订阅，当有新的任务被添加时，所有订阅了该事件的客户端都会立即收到通知。

1. 定义订阅类型

首先，我们需要在 schema.js 中定义订阅类型。修改 typeDefs，添加一个新的订阅字段 taskAdded，它会在有新任务添加时触发：

const { gql } = require('apollo-server');

const typeDefs = gql`
  type Task {
    id: ID!
    title: String!
    completed: Boolean!
  }

  type Query {
    tasks: [Task!]!
  }

  type Mutation {
    addTask(title: String!): Task!
    completeTask(id: ID!): Task!
  }

  type Subscription {
    taskAdded: Task!
  }
`;

module.exports = typeDefs;

Subscription 类型中的 taskAdded 字段表示每当有新任务被添加时，订阅者会收到一个 Task 对象作为通知。

2. 实现订阅解析器

接下来，我们需要为 taskAdded 实现解析器。在 resolvers.js 中，添加以下代码：

const { PubSub } = require('apollo-server');

const pubsub = new PubSub();
const TASK_ADDED = 'TASK_ADDED';

const resolvers = {
  Query: {
    tasks: () => tasks,
  },
  Mutation: {
    addTask: (_, { title }) => {
      const newTask = {
        id: Math.random().toString(36).substr(2, 9),
        title,
        completed: false,
      };
      tasks.push(newTask);

      // 发布新任务事件
      pubsub.publish(TASK_ADDED, { taskAdded: newTask });

      return newTask;
    },
    completeTask: (_, { id }) => {
      const task = tasks.find(t => t.id === id);
      if (!task) throw new Error('Task not found');
      task.completed = true;
      return task;
    },
  },
  Subscription: {
    taskAdded: {
      subscribe: () => pubsub.asyncIterator([TASK_ADDED]),
    },
  },
};

module.exports = resolvers;

这里我们引入了 PubSub，它是一个发布/订阅模式的实现，允许我们在事件发生时发布消息，并让订阅者接收到这些消息。TASK_ADDED 是我们定义的一个事件名称，每当有新任务被添加时，我们都会通过 pubsub.publish 发布该事件，并将新任务作为参数传递。

在 Subscription 解析器中，taskAdded 字段的 subscribe 方法返回了一个异步迭代器（async iterator），它会监听 TASK_ADDED 事件，并在事件发生时将新任务推送给订阅者。

3. 测试订阅

现在我们已经完成了订阅功能的实现，接下来可以测试一下效果。回到 GraphQL Playground，打开两个标签页，分别执行以下操作：

标签页 1：订阅新任务

在第一个标签页中，执行以下订阅查询：

subscription {
  taskAdded {
    id
    title
    completed
  }
}

你会看到 Playground 显示了一个订阅结果窗口，等待新任务的出现。

标签页 2：添加新任务

在第二个标签页中，执行以下变更操作，添加一个新任务：

mutation {
  addTask(title: "Real-time data is awesome!") {
    id
    title
    completed
  }
}

当你提交这个变更时，你会发现第一个标签页中的订阅结果窗口立即显示了新任务的信息！这意味着订阅功能已经成功实现了。👏

处理多个订阅者

在实际应用中，可能会有多个客户端同时订阅同一个事件。为了确保每个客户端都能正确接收到通知，我们需要确保订阅机制是线程安全的，并且能够处理并发请求。

幸运的是，PubSub 内置了对多订阅者的支持。每当有新事件发布时，PubSub 会自动将消息推送给所有订阅了该事件的客户端。因此，我们不需要做任何额外的工作来处理多个订阅者。

为了验证这一点，你可以打开更多的浏览器标签页，重复上述的订阅和变更操作。你会发现每个标签页都会收到相同的通知，证明我们的订阅功能确实支持多个客户端。

过滤订阅

有时候，你可能希望根据某些条件来过滤订阅的结果。例如，在一个多人协作的任务管理系统中，用户可能只想订阅与自己相关的任务更新，而不是所有的任务。

为了实现这一点，我们可以在订阅解析器中添加参数，并根据这些参数来过滤事件。修改 schema.js，为 taskAdded 添加一个 userId 参数：

const { gql } = require('apollo-server');

const typeDefs = gql`
  type Task {
    id: ID!
    title: String!
    completed: Boolean!
  }

  type Query {
    tasks: [Task!]!
  }

  type Mutation {
    addTask(title: String!, userId: ID!): Task!
    completeTask(id: ID!): Task!
  }

  type Subscription {
    taskAdded(userId: ID): Task!
  }
`;

module.exports = typeDefs;

接下来，修改 resolvers.js，在 addTask 解析器中传递 userId，并在 taskAdded 订阅解析器中根据 userId 进行过滤：

const tasks = [];
const pubsub = new PubSub();
const TASK_ADDED = 'TASK_ADDED';

const resolvers = {
  Query: {
    tasks: () => tasks,
  },
  Mutation: {
    addTask: (_, { title, userId }) => {
      const newTask = {
        id: Math.random().toString(36).substr(2, 9),
        title,
        completed: false,
        userId,
      };
      tasks.push(newTask);

      // 发布新任务事件，带上 userId
      pubsub.publish(TASK_ADDED, { taskAdded: newTask });

      return newTask;
    },
    completeTask: (_, { id }) => {
      const task = tasks.find(t => t.id === id);
      if (!task) throw new Error('Task not found');
      task.completed = true;
      return task;
    },
  },
  Subscription: {
    taskAdded: {
      subscribe: (parent, { userId }, context) => {
        // 只订阅与当前 userId 相关的任务
        return pubsub.asyncIterator([TASK_ADDED]).filter(
          ({ taskAdded }) => taskAdded.userId === userId
        );
      },
    },
  },
};

module.exports = resolvers;

在这段代码中，addTask 解析器现在接受一个 userId 参数，并将其存储在任务对象中。taskAdded 订阅解析器则使用 filter 方法，只将与指定 userId 匹配的任务推送给订阅者。

测试过滤订阅

为了测试过滤订阅的效果，你可以打开两个不同的浏览器标签页，分别订阅不同的 userId，然后添加一些任务。你会发现每个标签页只会收到与自己 userId 相关的任务通知，而不会收到其他用户的任务更新。

错误处理和优化

虽然我们已经实现了一个基本的订阅功能，但在实际应用中，还需要考虑一些常见的问题，比如错误处理和性能优化。

1. 错误处理

在订阅解析器中，可能会遇到各种错误，比如无效的参数、数据库连接失败等。为了确保订阅的健壮性，我们应该在解析器中添加适当的错误处理逻辑。

例如，我们可以在 taskAdded 解析器中捕获异常，并返回一个自定义的错误消息：

Subscription: {
  taskAdded: {
    subscribe: (parent, { userId }, context) => {
      try {
        return pubsub.asyncIterator([TASK_ADDED]).filter(
          ({ taskAdded }) => taskAdded.userId === userId
        );
      } catch (error) {
        throw new Error('Failed to subscribe to task updates');
      }
    },
  },
};

此外，Apollo Server 还提供了一些内置的错误处理机制，比如 formatError，可以用来统一处理所有 GraphQL 错误。你可以在 index.js 中配置 formatError，以便在生产环境中返回更加友好的错误信息：

const server = new ApolloServer({
  typeDefs,
  resolvers,
  formatError: error => {
    console.error(error);
    return {
      message: 'An unexpected error occurred',
      code: 'INTERNAL_SERVER_ERROR',
    };
  },
});

2. 性能优化

随着订阅者的数量增加，服务器的负载也会逐渐增大。为了提高性能，我们可以采取一些优化措施，比如：

• 批量处理事件：如果短时间内有大量的事件发生，可以考虑将这些事件批量处理，减少不必要的通知推送。
• 限制订阅频率：对于一些高频更新的事件，可以设置一个最小的时间间隔，避免过度频繁的通知。
• 使用缓存：对于一些不经常变化的数据，可以使用缓存来减少数据库查询的次数。

此外，还可以考虑使用更高效的 WebSocket 库，比如 subscriptions-transport-ws，它专门为 GraphQL 订阅设计，提供了更好的性能和稳定性。

结语

经过今天的讲座，相信你已经掌握了如何使用 Apollo Server 开发 GraphQL 订阅，实现真正的实时数据推送。通过订阅功能，你可以轻松构建出具有实时交互能力的应用程序，提升用户体验。

当然，订阅只是一个开始，GraphQL 还有许多其他强大的功能等待你去探索。希望今天的讲座能够为你提供一些启发，帮助你在未来的项目中更好地利用 GraphQL 的优势。

如果你有任何问题或想法，欢迎在评论区留言讨论！期待与你一起继续学习和进步。😊

祝你编码愉快，再见！👋