java - 关闭 AsynchronousSocketChannel 时连接重置错误

Question

我正在 Kotlin 中试验一个简单的 HTTP1.1 服务器，以便尝试重现一个更复杂的问题，即在关闭 AsynchronousSocketChannel 时内容似乎被切断

net::ERR_CONTENT_LENGTH_MISMATCH 200(正常)

首先是一个使调试更容易的辅助函数:

@Suppress("NOTHING_TO_INLINE")
inline fun ByteArray.debug(): String = this.map {
    when (it) {
        '\t'.code.toByte() -> "\\t"
        '\r'.code.toByte() -> "\\r"
        '\n'.code.toByte() -> "\\n\n"
        else -> "${it.toInt().toChar()}"
    }
}.toTypedArray().joinToString(separator = "") { it }

这里是整个简化的 Web 服务器及其所有导入，您应该能够将其复制并粘贴到 .kt 文件中以运行:

import java.net.InetSocketAddress
import java.nio.ByteBuffer
import java.nio.channels.AsynchronousServerSocketChannel
import java.nio.channels.AsynchronousSocketChannel
import java.nio.channels.CompletionHandler
import java.util.concurrent.TimeUnit

fun main() {

    val len = 1_000_000

    val server =
        AsynchronousServerSocketChannel.open().bind(InetSocketAddress(5555))

    while (true) {
        val channel = server.accept().get()

        val body = "#".repeat(len)
        val headers = mutableListOf<String>()
        headers.add("HTTP/1.1 200 OK")
        headers.add("Server: Test Server")
        headers.add("Content-Type: text/plain")
        headers.add("Connection: close")
        headers.add("Content-Length: ${body.length}")
        val header = headers.joinToString(separator = "\r\n") + "\r\n\r\n"

        println("==== header (size=${header.toByteArray().size})")
        println(header.toByteArray().debug())
        println("==== body (size=${body.toByteArray().size})")
        val data = "$header$body".toByteArray()
        println("==== data (size=${data.size})")

        println(data.debug())
        channel.write(
            ByteBuffer.wrap(data),
            30,
            TimeUnit.SECONDS,
            channel,
            object : CompletionHandler<Int?, AsynchronousSocketChannel> {
                override fun completed(result: Int?, channel: AsynchronousSocketChannel) {
                    println(result)
                    channel.close()
                }

                override fun failed(exc: Throwable?, channel: AsynchronousSocketChannel) {
                    channel.close()
                }
            })

    }

}

运行它并在 localhost:5555 打开浏览器，迎接我的是连接重置

查看浏览器网络控制台，我可以看到响应 header 看起来是正确的:

查看输出，我可以看到它也是正确的并且与我们在浏览器网络控制台中看到的内容相匹配，并且末尾的 1000110 打印在 Completion Handler 中，匹配数据的总大小，但没有呈现任何内容，浏览器提示连接重置。

==== header (size=110)
HTTP/1.1 200 OK\r\n
Server: Test Server\r\n
Content-Type: text/plain\r\n
Connection: close\r\n
Content-Length: 1000000\r\n
\r\n

==== body (size=1000000)
==== data (size=1000110)
HTTP/1.1 200 OK\r\n
Server: Test Server\r\n
Content-Type: text/plain\r\n
Connection: close\r\n
Content-Length: 1000000\r\n
\r\n
#####################################################################################.......
1000110

如果我在 channel.close() 之前添加一个 Thread.sleep，它会正常工作，但显然浏览器会等待整整一秒才能再次连接，所以这肯定是不是解决方案。

channel.write(
    ByteBuffer.wrap(data),
    30,
    TimeUnit.SECONDS,
    channel,
    object : CompletionHandler<Int?, AsynchronousSocketChannel> {
        override fun completed(result: Int?, channel: AsynchronousSocketChannel) {
            println(result)
            Thread.sleep(1000)
            channel.close()
        }

        override fun failed(exc: Throwable?, channel: AsynchronousSocketChannel) {
            channel.close()
        }
    })

建议在调用 close() 之前调用 channel.shutdownOutput() 和 channel.read() 的响应之一

override fun completed(result: Int?, channel: AsynchronousSocketChannel) {
    println(result)
    channel.shutdownOutput()
    channel.read(ByteBuffer.allocate(1)).get()
    channel.close()
}

如果我使用 allocate(1)，它不能解决问题，但 allocate(very-big-number) 确实有效，但实际上没有不同于在这里调用 Thread.sleep。

如果我将它部署到 AWS，并在负载均衡器后面使用一个短的 Thread.sleep，我会遇到net::ERR_CONTENT_LENGTH_MISMATCH 200 (OK) 这意味着它正在写入一些数据，但是在负载均衡器可以读取所有数据之前流被切断，实际上与 net::ERR_CONNECTION_RESET 200(确定) 错误。

关闭 AsynchronousSocketChannel 而不会在浏览器中遇到连接重置错误或内容长度不匹配错误的正确方法是什么？

编辑:这是更完整的演示，我仍然可以在其中重现错误。在此示例中，我首先阅读请求，然后再编写响应。为了使其更具可读性，我将完成处理程序包装在 suspendCoroutine 中，这样我就可以调用 readingSuspending 和 writeSuspending。

import kotlinx.coroutines.*
import java.net.InetSocketAddress
import java.nio.ByteBuffer
import java.nio.channels.AsynchronousServerSocketChannel
import java.nio.channels.AsynchronousSocketChannel
import java.nio.channels.CompletionHandler
import java.util.concurrent.TimeUnit
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
import kotlin.coroutines.suspendCoroutine
import kotlin.time.Duration
import kotlin.time.Duration.Companion.seconds

object Test {

    val len = 1_000_000_000

    suspend fun AsynchronousServerSocketChannel.acceptSuspending() =
        suspendCoroutine<AsynchronousSocketChannel> { continuation ->
            this.accept(
                null, object : CompletionHandler<AsynchronousSocketChannel, Nothing?> {
                    override fun completed(result: AsynchronousSocketChannel, attachment: Nothing?) {
                        continuation.resume(result)
                    }

                    override fun failed(exc: Throwable, attachment: Nothing?) {
                        continuation.resumeWithException(exc)
                    }
                })
        }

    suspend fun AsynchronousSocketChannel.writeSuspending(
        buffer: ByteBuffer,
        timeout: Duration = 60.seconds,
        closeWhenDone: Boolean = false,
    ) = suspendCoroutine<Int> { continuation ->
        this.write(buffer, timeout.inWholeSeconds, TimeUnit.SECONDS, null, object : CompletionHandler<Int, Nothing?> {
            override fun completed(size: Int, attachment: Nothing?) {
                continuation.resume(size)
            }

            override fun failed(exc: Throwable, attachment: Nothing?) {
                continuation.resumeWithException(exc)
            }
        })
    }

    suspend fun AsynchronousSocketChannel.readSuspending(
        buffer: ByteBuffer,
        timeout: Duration = 5.seconds,
    ) = suspendCoroutine<Int> { continuation ->
        this.read(buffer, timeout.inWholeSeconds, TimeUnit.SECONDS, null, object : CompletionHandler<Int, Nothing?> {
            override fun completed(size: Int, attachment: Nothing?) {
                continuation.resume(size)
            }

            override fun failed(exc: Throwable, attachment: Nothing?) {
                continuation.resumeWithException(exc)
            }
        }
        )
    }

    @JvmStatic
    fun main(args: Array<String>) = runBlocking(Dispatchers.Default) {

        val server = withContext(Dispatchers.IO) {
            AsynchronousServerSocketChannel.open().bind(InetSocketAddress(5555))
        }

        while (true) {
            val channel = server.acceptSuspending()
            supervisorScope {
                launch {
                    val buffer = ByteBuffer.allocate(1000)

                    // reading
                    do {
                        val size = channel.readSuspending(buffer.rewind(), 30.seconds)
                        println(String(buffer.array().sliceArray(0..size)))
                    } while (!buffer.hasRemaining())

                    // build response
                    val body = "#".repeat(len)
                    val headers = mutableListOf<String>()
                    headers.add("HTTP/1.1 200 OK")
                    headers.add("Server: Test Server")
                    headers.add("Content-Type: text/plain")
                    headers.add("Content-Length: ${body.length}")
                    val header = headers.joinToString(separator = "\r\n") + "\r\n\r\n"
                    val data = "$header$body".toByteArray()

                    // writing
                    channel.writeSuspending(ByteBuffer.wrap(data), 30.seconds)
                    withContext(Dispatchers.IO) {
                        channel.close()
                    }
                }
            }
        }
    }
}
 

Answer 1

不是套接字程序员，只是一些想法...

代码实际上是同步的，所有逻辑都需要在 CompletionHandler 中实现，但我相信它证明了这个问题。

public static void main(String[] args) throws Exception {
    int len = 50_000_000;

    AsynchronousServerSocketChannel server =
            AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(5555));

    while (true) {
        try (AsynchronousSocketChannel channel = server.accept().get()) {
            // reading client headers, without that mitigation
            // of the race condition at the bottom of loop fails
            // one the other hand, I failed to reproduce the
            // problem when bottom read is commented out 
            System.out.println("Header: " + channel.read(ByteBuffer.allocate(4096)).get());
            StringBuilder body = new StringBuilder("HTTP/1.1 200 OK\r\n" +
                    "server: Cloud Auctioneers\r\n" +
                    "content-type: text/plain\r\n" +
                    // tell client to close socket
                    "connection: close\r\n" +
                    "content-length: " + len + "\r\n\r\n");
            for (int i = 0; i < len; i++) {
                body.append('#');
            }
            ByteBuffer buff = ByteBuffer.wrap(body.toString().getBytes());
            // according to javadoc write method does not guarantee
            // that it will send the all data to client, at least
            // without this loop client receives just 256K on my laptop
            while (buff.hasRemaining()) {
                Integer written = channel.write(buff).get();
                System.out.println("Written: " + written);
                // not sure here about null
                if (written == null) {
                    break;
                }
            }
            // here we are trying to mitigate race condition between try-with-resources
            // and delivering pending data. according to man 2 send it sends data
            // asynchronously and there is no way to understand whether that was
            // successful or not - trying to read something from socket
            System.out.println("Footer: " + channel.read(ByteBuffer.allocate(4096)).get());
        }
    }
}

---

更新。对该主题进行了一些研究，以下是一些想法:

1. 在写入套接字的情况下，代码片段如下:

public <A> void completed(Integer result, A attachment) {
    if (result < 0) {
        // socket closed
        channel.close();
        return;
    }
    if (buffer.hasRemaining()) {
        channel.write(buffer, null, this);
    }
} 

由于以下原因似乎是强制性的:

• 似乎 java.nio 没有将所有错误路由到 CompletionHandler#failed，不知道为什么，这就是我观察到的
• 它不会向客户端发送完整的缓冲区 - 这可能是合理的，因为它始终是一个问题，该怎么做:compact 缓冲区并通过未决数据填充它或向客户端发送提醒，但是我更愿意在 AsynchronousSocketChannel#write
中有一个定义行为的标志

1. 关于谁负责关闭连接的问题 - 答案取决于 HTTP 协议(protocol)的版本(例如客户端):

• 在 HTTP/1.1 的情况下，服务器可以通过发送 HTTP header 请求客户端关闭连接 Connection: close

• 如果是HTTP/1.0服务器必须自己关闭连接

  从基础架构的角度来看，这两种情况之间的区别如下:启动关闭的一方在 TIME_WAIT 状态下获取套接字，并且在服务器的情况下这是不希望的，因此我们还需要设置 SO_REUSEADDR 为 true(AsynchronousSocketChannel 不支持 SO_LINGER)

因此，实现最终取决于 HTTP 协议(protocol)的版本，这里我更愿意使用现有的库，如 netty 而不是解决那些 http 和 java.nio 谜题，不过大意是:

• 读取 HTTP header 直到 '\r\n\r\n' 或 '\n\n'，确定客户端版本
• 跟踪写入客户端的数据量并重申
• 如果 IO 返回 -1 或抛出异常则关闭 channel
• 如果客户端是 HTTP/1.1 发送 Connection: close 并等待 -1
• 如果客户端是 HTTP/1.0 关闭 channel