Java中netty怎样主动推送信息。

Posted 2023-03-26

我需要用netty进行长连接，现在已经做到，当客户端发送给我消息，我接收到并做一定的处理后返回一定消息。
但我现在想做到的是能够外部控制server发送的内容，比如：“向所有客户端发送某消息”，或者：“向某个客户端发送消息”，我大概查了一下，网上说要用map存下channel，但是在下列代码里，应当存哪个channel呢？存到后，就可以用channel进行writeandFlash进行发送吗？

客户端是发送消息给服务器的，服务器不可能主动发消息给客户端。就是长链接，比如，客户端发送一个消息你好的消息给服务器，那么服务器会接收到，这时候就有客户端的所有信息，比如id+ip组成的管道，管道id（一般是用户id）。
如果“向所有客户端发送某消息”，那么要想办法在设置一个资格（比如登陆的时候保存客户端的信息用对象存储，这样更利于更改更新管道）。然后把所有客户端放到一个list里面存储。遍历list，给每个用户发送消息。 参考技术A 你看web socket的时候没发现页面都是要求html5吗，这个适用性当前环境下不强，而且这方面的具体应用其实不多，使用ajax刷新简单方便，兼容性也好，基本上满足一般的需求。 参考技术B 存ctx.channel()本回答被提问者采纳

Netty 长连接 Server 主动推送功能实现及问题

背景

使用 Netty 实现长连接，Client 端主动发起请求到 Server 端后，Server 缓存 Channel ，后续业务需要主动向 Client 推送数据时，直接遍历 Channel 写入数据。

用 Netty 实现 Http 服务器还是挺方便的，对 Netty 不太熟悉的时候，看着各种类似的代码，都能用，没有细究添加到 Pipeline 的处理器及区别。用了一个 HttpContentCompressor ，导致主动推送 HttpResponse 数据时写入失败。

本文将记录该问题的始末。

常规 Netty HttpServer 的代码

Netty 作为服务端，构建代码很常见：

public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel channel) {
        channel.pipeline().addLast(new HttpServerCodec());
        // 提供在握手时聚合 HttpRequest，设置最大长度为 100Mb:超长则会报 413Request Entity Too Large
        channel.pipeline().addLast(new HttpObjectAggregator(104857600));
        // 压缩返回数据 HttpResponse
        channel.pipeline().addLast(new HttpContentCompressor());
        channel.pipeline().addLast(new ChunkedWriteHandler());
        channel.pipeline().addLast(new MyServerHandler());
    }
}

服务端与客户端保持长连接，必要的时候通过 Channel 向 Client 主动推送数据。上面的代码，在主动推送数据时，会报异常。

Netty Server 端主动向通道写入异常

Server 端的业务需要通过 Channel 向 Client 推送数据，推送代码为：

FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer(result, CharsetUtil.UTF_8));   //  Unpooled.wrappedBuffer(responseJson)
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8");       // HttpHeaderValues.TEXT_PLAIN.toString()
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
if (channel.isWritable()) {
    channel.writeAndFlush(httpResponse).addListener((ChannelFutureListener) future -> {
        if (future.isSuccess()) {
            logger.info("回写成功");
        } else {
            logger.error("回写失败",future.cause());
        }
    });
}

在使用了 HttpContentCompressor 后，writeAndFlush 回调总是进入异常分支：

完整的异常信息：

io.netty.handler.codec.EncoderException: java.lang.IllegalStateException: 
cannot send more responses than requests
        at io.netty.handler.codec.
        MessageToMessageEncoder.write(MessageToMessageEncoder.java:107)

去掉 HttpContentCompressor

去掉 HttpContentCompressor 后Server 主动推送的数据能够正确发送，为什么这个压缩 HttpResponse 的处理器会对数据写入有这么大的影响呢？

启示录

最初以为 Netty 实现的是 http 服务，Client 发送请求通过 Http 协议完成的，Server 主动推送数据失败是因为 Http 协议的局限性导致的。

于是尝试用 WebSocket 协议，这样反向推送是 OK 的。但是 Client 端全程只用一个 Socket 连接，导致其他正常的 Http 请求都升级为 WebSocket 请求了，其他 Http 协议处理流程无法正常运行。

结论：

1. 如果要使用 HttpContentCompressor ，那么反向数据推送只能通过 WebSocket 协议进行。Client 端就需要区分 Http 请求和 Websocket 请求，用不同的 Socket 进行数据发送。
2. 去掉 HttpContentCompressor 类，Client 采用 Http 请求缓存的 Channel 能正常推送数据，但是响应 Body 数据不会有压缩功能。应用中没有大数据量的传输业务，所以去掉也没啥影响。