Swift 6下NIO Mock服务器并发错误：非隔离上下文调用Actor方法

核心问题分析 #

MockHTTPServer是actor类型，其getResponse(for:)方法默认是actor隔离的，仅能在actor内部同步调用。但MockHTTPHandler是非隔离的ChannelInboundHandler，在同步的channelRead方法中直接调用该方法，违反了Swift6的并发安全规则。直接用Task包裹后触发竞态错误，原因是NIO的ChannelHandlerContext并非线程安全，异步Task中访问它可能在非EventLoop线程执行，引发线程竞争。

简单解决步骤 #

1. 修改actor的响应查询方法为异步：
   将getResponse(for:)改为异步方法，允许非actor上下文通过await调用：

   func getResponse(for path: String) async -> (HTTPResponseStatus, String)? {
       return responses[path]
   }
   

2. 在channelRead中异步查询并切回EventLoop执行响应：
   在channelRead中用Task异步调用actor的方法，拿到结果后切回Channel对应的EventLoop执行响应操作（保证NIO线程安全）：

   func channelRead(context: ChannelHandlerContext, data: NIOAny) {
       let reqPart = unwrapInboundIn(data)
       
       guard case .head(let head) = reqPart else { return }
       
       guard let server = self.server else {
           respond(context: context, status: .internalServerError, body: "Server not found")
           return
       }
       
       // 用Task异步查询响应，避免阻塞EventLoop
       Task { [weak context, head] in
           guard let context = context else { return }
           // 异步调用actor的隔离方法
           let response = await server.getResponse(for: head.uri)
           // 切回EventLoop执行响应操作，保证NIO线程安全
           context.eventLoop.execute {
               if let (status, body) = response {
                   self.respond(context: context, status: status, body: body)
               } else {
                   self.respond(context: context, status: .notFound, body: "Not Found")
               }
           }
       }
   }
   

修改后的完整关键代码片段 #

fileprivate actor MockHTTPServer {
    private let group: EventLoopGroup
    private let channel: Channel
    let port: Int
    
    private var responses: [String: (HTTPResponseStatus, String)] = [:]
    
    private init(group: EventLoopGroup, channel: Channel) {
        self.group = group
        self.channel = channel
        self.port = channel.localAddress?.port ?? -1
    }
    
    static func start(group: EventLoopGroup) async throws -> MockHTTPServer {
        let server = try await withCheckedThrowingContinuation { continuation in
            let bootstrap = ServerBootstrap(group: group)
                .serverChannelOption(ChannelOptions.backlog, value: 256)
                .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
                .childChannelInitializer { channel in
                    channel.pipeline.configureHTTPServerPipeline().flatMap { _ in
                        let handler = MockHTTPHandler()
                        return channel.pipeline.addHandler(handler)
                    }
                }
            
            bootstrap.bind(host: "localhost", port: 0).whenComplete { result in
                switch result {
                case .success(let channel):
                    let server = MockHTTPServer(group: group, channel: channel)
                    channel.pipeline.handler(type: MockHTTPHandler.self).whenSuccess { handler in
                        handler.server = server
                    }
                    continuation.resume(returning: server)
                case .failure(let error):
                    continuation.resume(throwing: error)
                }
            }
        }
        
        return server
    }
    
    func stop() async throws {
        try await channel.close()
    }
    
    func addResponse(for path: String, statusCode: HTTPResponseStatus, body: String) {
        responses[path] = (statusCode, body)
    }
    
    func getResponse(for path: String) async -> (HTTPResponseStatus, String)? {
        return responses[path]
    }
    
    private final class MockHTTPHandler: ChannelInboundHandler {
        typealias InboundIn = HTTPServerRequestPart
        typealias OutboundOut = HTTPServerResponsePart
        
        weak var server: MockHTTPServer?
        
        func channelRead(context: ChannelHandlerContext, data: NIOAny) {
            let reqPart = unwrapInboundIn(data)
            
            guard case .head(let head) = reqPart else { return }
            
            guard let server = self.server else {
                respond(context: context, status: .internalServerError, body: "Server not found")
                return
            }
            
            Task { [weak context, head] in
                guard let context = context else { return }
                let response = await server.getResponse(for: head.uri)
                context.eventLoop.execute {
                    if let (status, body) = response {
                        self.respond(context: context, status: status, body: body)
                    } else {
                        self.respond(context: context, status: .notFound, body: "Not Found")
                    }
                }
            }
        }
        
        private func respond(context: ChannelHandlerContext, status: HTTPResponseStatus, body: String) {
            var headers = HTTPHeaders()
            headers.add(name: "Content-Type", value: "application/json")
            headers.add(name: "Content-Length", value: "\(body.utf8.count)")
            
            let head = HTTPResponseHead(version: .http1_1, status: status, headers: headers)
            context.write(wrapOutboundOut(.head(head)), promise: nil)
            
            var buffer = context.channel.allocator.buffer(capacity: body.utf8.count)
            buffer.writeString(body)
            context.write(wrapOutboundOut(.body(.byteBuffer(buffer))), promise: nil)
            context.writeAndFlush(wrapOutboundOut(.end(nil)), promise: nil)
        }
    }
}

为什么这样能解决问题 #

• 将getResponse改为异步方法后，非actor上下文可以通过await安全调用actor的隔离方法，符合Swift6的并发规则。
• 通过context.eventLoop.execute将响应操作切回Channel的EventLoop线程，保证NIO的Channel操作始终在正确的线程执行，避免了线程竞态问题。

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