10年架构师教你使用Netty实现数据采集服务器

使用Netty实现数据采集服务器

为了更好地讲解如何结合NIO和异步编程以实现支持高并发、高性能的数据采集服务器，本节采用Netty框架重构数据采集服务器。Netty是一个基于NIO的异步事件驱动网络应用框架，用于快速开发具有可维护性的高性能协议服务器和客户端。Netty中的一些概念和设计模式有些复杂，但是非常有用，初次接触Netty的读者还是需要对其有所了解。

可以说，Netty把NIO和异步编程的哲学发挥到了淋漓尽致。在Netty中，几乎所有涉及网络操作的地方均采用异步回调的方式。图2-12展示了Netty的工作原理。首先，Netty用reactor线程监听ServerSocketChannel，每个ServerSocketChannel对应一个实际的端口。如果需要监听多个端口，则需要为reactor线程池配置多个线程。

当reactor线程监听的ServerSocketChannel监测到连接请求事件（OP_ACCEPT）时，就为接收到的连接套接字建立一个SocketChannel，并将该SocketChannel委托给工作线程池中的某个工作线程做后续处理。之后，当工作线程监测到SocketChannel上有数据可读（OP_READ）时，就调用相关的回调句柄（handler）对数据进行读取和处理，并返回最终的处理结果。另外，在Netty相关代码中，通常将reactor线程池称为boss group，而将工作线程池称为work group，大家在阅读Netty相关代码时知晓这两个概念即可。

图2-12 Netty的工作原理

使用Netty实现数据采集API

下面我们使用Netty来实现数据采集服务器。

public static void main(String[] args) {

final int port = 8081;

final EventLoopGroup bossGroup = new NioEventLoopGroup(1);

final EventLoopGroup workerGroup = new NioEventLoopGroup();

try {

final ServerBootstrap bootstrap = new ServerBootstrap()

.group(bossGroup, workerGroup)

.channel(NioServerSocketChannel.class)

.childHandler(new ServerInitializer())

.option(ChannelOption.SO_BACKLOG, 1024);

final ChannelFuture f = bootstrap.bind(port).sync();

logger.info(String.format(\”NettyDataCollector: running on port[%d]\”,

port));

f.channel().closeFuture().sync();

} catch (final InterruptedException e) {

logger.error(\”NettyDataCollector: an error occurred while running\”, e);

} finally {

bossGroup.shutdownGracefully();

workerGroup.shutdownGracefully();

}

}

public class ServerInitializer extends ChannelInitializer<SocketChannel> {

private static final int MAX_CONTENT_LENGTH = 1024 * 1024;

@Override

public void initChannel(SocketChannel ch) throws Exception {

ch.pipeline().addLast(\”http-codec\”, new HttpServerCodec());

ch.pipeline().addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH));

ch.pipeline().addLast(\”handler\”, new ServerHandler());

}

}

public class ServerHandler extends

SimpleChannelInboundHandler<HttpRequest> {

private static final Logger logger = LoggerFactory.getLogger(NettyData

Collector.class);

private final String kafkaBroker = \”127.0.0.1:9092\”;

private final String topic = \”collector_event\”;

private final KafkaSender kafkaSender = new KafkaSender(kafkaBroker);

private JSONObject doExtractCleanTransform(JSONObject event) {

// TODO: 实现抽取、清洗、转化具体逻辑

return event;

}

@Override

protected void channelRead0(ChannelHandlerContext ctx, HttpRequest req)

throws Exception {