在Next.js API中配置带WebSocket的GraphQL Yoga

核心思路 #

Next.js的API路由默认是请求响应模型，但WebSocket需要长连接，所以得结合Yoga的特性，分两种主流路由方案实现：

方案一：App Router（Next.js 13+ 推荐） #

在app/api/graphql/route.ts创建路由，同时处理HTTP和WebSocket请求：

import { createYoga, createSchema } from 'graphql-yoga'
import { useServer } from 'graphql-ws/lib/use/ws'
import { WebSocketServer } from 'ws'

// 定义你的GraphQL Schema与Resolver
const schema = createSchema({
  typeDefs: /* GraphQL */ `
    type Query {
      hello: String!
    }
    type Subscription {
      count: Int!
    }
  `,
  resolvers: {
    Query: {
      hello: () => 'Hello World!'
    },
    Subscription: {
      count: {
        subscribe: async function* () {
          let count = 0
          while (true) {
            yield { count: count++ }
            await new Promise(resolve => setTimeout(resolve, 1000))
          }
        }
      }
    }
  }
})

const yoga = createYoga({
  schema,
  graphiql: {
    subscriptionsProtocol: 'WS'
  },
  fetchAPI: { Response } // 适配Next.js的响应对象
})

// 处理HTTP GET/POST请求
export const GET = yoga
export const POST = yoga

// 禁用Next.js默认的body解析，交给Yoga处理
export const config = {
  api: {
    bodyParser: false
  }
}

// 仅在服务器端启动WebSocket服务
if (typeof window === 'undefined') {
  const server = yoga.getServer()
  const wsServer = new WebSocketServer({
    server,
    path: yoga.graphqlEndpoint
  })

  // 关联Yoga与WebSocket服务器
  useServer(
    {
      execute: (args: any) => args.rootValue.execute(args),
      subscribe: (args: any) => args.rootValue.subscribe(args),
      onSubscribe: async (ctx, msg) => {
        const { schema, execute, subscribe, contextFactory, parse, validate } =
          yoga.getEnveloped(ctx)

        const args = {
          schema,
          operationName: msg.payload.operationName,
          document: parse(msg.payload.query),
          variableValues: msg.payload.variables,
          contextValue: await contextFactory(),
          rootValue: {
            execute,
            subscribe
          }
        }

        const errors = validate(args.schema, args.document)
        if (errors.length) return errors
        return args
      }
    },
    wsServer
  )
}

方案二：Pages Router（Next.js 12及以下） #

在pages/api/graphql.ts创建API路由，手动处理WebSocket升级请求：

import { createYoga, createSchema } from 'graphql-yoga'
import type { NextApiRequest, NextApiResponse } from 'next'
import { WebSocketServer } from 'ws'
import { useServer } from 'graphql-ws/lib/use/ws'

// 定义Schema与Resolver
const schema = createSchema({
  typeDefs: /* GraphQL */ `
    type Query {
      hello: String!
    }
    type Subscription {
      count: Int!
    }
  `,
  resolvers: {
    Query: {
      hello: () => 'Hello World!'
    },
    Subscription: {
      count: {
        subscribe: async function* () {
          let count = 0
          while (true) {
            yield { count: count++ }
            await new Promise(resolve => setTimeout(resolve, 1000))
          }
        }
      }
    }
  }
})

const yoga = createYoga({
  schema,
  graphiql: {
    subscriptionsProtocol: 'WS'
  },
  graphqlEndpoint: '/api/graphql'
})

// 全局存储WebSocket实例，避免重复创建
let wsServer: WebSocketServer | null = null

export default async function handler(req: NextApiRequest, res: NextApiResponse) {
  // 处理普通HTTP请求
  if (req.method !== 'GET' || !req.headers.upgrade) {
    return yoga(req, res)
  }

  // 初始化WebSocket服务器
  if (!wsServer) {
    const server = res.socket?.server
    if (!server) {
      res.status(500).send('服务器不可用')
      return
    }

    wsServer = new WebSocketServer({
      server,
      path: '/api/graphql'
    })

    // 关联Yoga与WebSocket
    useServer(
      {
        execute: (args: any) => args.rootValue.execute(args),
        subscribe: (args: any) => args.rootValue.subscribe(args),
        onSubscribe: async (ctx, msg) => {
          const { schema, execute, subscribe, contextFactory, parse, validate } =
            yoga.getEnveloped(ctx)

          const args = {
            schema,
            operationName: msg.payload.operationName,
            document: parse(msg.payload.query),
            variableValues: msg.payload.variables,
            contextValue: await contextFactory(),
            rootValue: {
              execute,
              subscribe
            }
          }

          const errors = validate(args.schema, args.document)
          if (errors.length) return errors
          return args
        }
      },
      wsServer
    )
  }

  // 处理WebSocket升级请求
  res.socket?.on('upgrade', (request, socket, head) => {
    wsServer?.handleUpgrade(request, socket, head, (ws) => {
      wsServer?.emit('connection', ws, request)
    })
  })
}

// 禁用body解析
export const config = {
  api: {
    bodyParser: false
  }
}

关键注意事项 #

• 安装依赖：确保安装@graphql-yoga/node、graphql、ws、graphql-ws这几个包
• 开发环境：Next.js热重载可能导致WebSocket服务重启，生产环境更稳定
• 部署限制：Vercel Serverless函数不支持WebSocket，需改用Edge Functions或部署到支持长连接的平台（如EC2、Render等）

内容的提问来源于stack exchange，提问作者laventnousportera