java - ScheduledFuture 仅运行 14 次

Question

我有一个非常奇怪的错误(或者可能是我的错误)。我在 Netty 上有非常简单的 HTTP 客户端和服务器。

客户端应该每隔 X 秒连接到服务器并发送带有一些 Cookie 的 GET 请求。目前仅此而已。

客户端执行此操作14次后停止执行。我尝试调试该问题，发现它在

之后停止

ChannelFuture channelFuture = clientNio.getBootstrap().connect().sync();
System.out.println("Connected!");

卡住了，什么也没发生。

客户端代码片段:

public class Test{

public static int count = 0;


public static void main(String[] args) throws Exception {

    /** Start NIO client:
     * 1. Create new session to the server
     * 2. Create new client NIO and pass the session in it
     * 3. Start client NIO
     * */
    NetworkSession session = new NetworkSession(new InetSocketAddress("127.0.0.1",8080));
    ClientNio clientNio = new ClientNio(session);
    clientNio.start();



    // This is URL for server (should take from session)
    String URL = System.getProperty("url", "http://127.0.0.1:8080/");
    URI uri = new URI(URL);
    String scheme = uri.getScheme() == null? "http" : uri.getScheme();
    String host = uri.getHost() == null? "127.0.0.1" : uri.getHost();
    int port = uri.getPort();
    String rawPath = uri.getRawPath();

    /** Because we use Netty it is logical to use it's EvenLoopGroup to schedule connection with the server each X seconds*/
    ScheduledFuture<?> future = clientNio.getGroup().scheduleAtFixedRate(
            new Runnable() {
                @Override
                public void run() {
                    try {
                        //  Make the connection attempt
                        ChannelFuture channelFuture = clientNio.getBootstrap().connect().sync();
                        System.out.println("Connected!");
                        Channel channel = channelFuture.channel();
                        System.out.println("Get channel " + channel);

                        // After we get connection we update our session state
                        session.setSessionChannel(channel);
                        session.setInternalAddress((InetSocketAddress) channel.localAddress());
                        // And get session metadata to be send to the server
                        String sessionMetaDataEncoded = Base64.encodeToString(session.getMetadata());

                        // Prepare the HTTP request
                        HttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, rawPath);
                        //request.headers().set(request.headers().set(HttpHeaderNames.HOST, host));
                        request.headers().set(HttpHeaderNames.HOST, host);
                        request.headers().set(HttpHeaderNames.CONNECTION,  HttpHeaderValues.CLOSE);
                        request.headers().set("Cookie", ClientCookieEncoder.STRICT.encode("sesionID",sessionMetaDataEncoded));


                        // Send the HTTP request.
                        channel.writeAndFlush(request);

                        // Wait for the server to close the connection.
                        channel.closeFuture().sync();
                        System.out.println("closed");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(count);
                    count++;
                }
            }, 0, 1, TimeUnit.SECONDS);
}

}

客户端NIO:

public class ClientNio {

private final InetSocketAddress serverAddress;
private final NetworkSession session;
private Bootstrap bootstrap;
private EventLoopGroup group;


//GETTERS
public Bootstrap getBootstrap() {
    return bootstrap;
}
public EventLoopGroup getGroup() {return group;}

public ClientNio(NetworkSession session) {
    this.session = session;
    this.serverAddress = this.session.getServerAddress();
}


public void start() {
    group = new NioEventLoopGroup();
    bootstrap = new Bootstrap();
    bootstrap.group(group)
            .channel(NioSocketChannel.class)
            .remoteAddress(serverAddress)
            .handler(new HTTPClientPipelineInitializer(session));
}

public void stop() {
    try {
        group.shutdownGracefully().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

}

客户端管道初始化程序

public class HTTPClientPipelineInitializer extends ChannelInitializer<SocketChannel> {

private final NetworkSession session;

public HTTPClientPipelineInitializer(NetworkSession session) {
    this.session = session;
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    pipeline.addLast("codec", new HttpClientCodec());
    /** Adds HttpObjectAggregator with a max message size of 512 KB to the pipeline */
    pipeline.addLast("aggregator", new HttpObjectAggregator(512 * 1024));
}

}

附注我发现它在 ScheduledFutureTask 类中停止，这是第 15 步的异常。

io.netty.util.concurrent.BlockingOperationException:DefaultChannelPromise@1919f0ff(incomplete)

如何处理这个问题？

Answer 1

经过一些调试后，我明白了这个问题。因为我们在同步期间使用 EvenLoop，所以我可能会被阻塞，这将导致

当用户在事件循环线程中执行阻塞操作时引发 IllegalStateException。如果在事件循环线程中执行阻塞操作，则阻塞操作很可能会进入死锁状态，从而抛出此异常。

因此，在我的情况下，我选择了简单的 ScheduledFuture future = Scheduler.scheduleAtFixedRate，一切正常。