Flink运行时之基于Netty的网络通信上

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 概述 本文以及接下来的几篇文章将介绍Flink运行时TaskManager间进行数据交换的核心部分——基于Netty通信框架远程请求ResultSubpartition。作为系列文章的第一篇,先列出一些需要了解的基础对象。

概述

本文以及接下来的几篇文章将介绍Flink运行时TaskManager间进行数据交换的核心部分——基于Netty通信框架远程请求ResultSubpartition。作为系列文章的第一篇,先列出一些需要了解的基础对象。

NettyConnectionManager

Netty连接管理器(NettyConnectionManager)是连接管理器接口(ConnectionManager)针对基于Netty的远程连接管理的实现者。它是TaskManager中负责网络通信的网络环境对象(NetworkEnvironment)的核心部件之一。

一个TaskManager中可能同时运行着很多任务实例,有时某些任务需要消费某远程任务所生产的结果分区,有时某些任务可能会生产结果分区供其他任务消费。所以对一个TaskManager来说,其职责并非单一的,它既可能充当客户端的角色也可能充当服务端角色。因此,一个NettyConnectionManager会同时管理着一个Netty客户端(NettyClient)和一个Netty服务器(NettyServer)实例。当然除此之外还有一个Netty缓冲池(NettyBufferPool)以及一个分区请求客户端工厂(PartitionRequestClientFactory,用于创建分区请求客户端PartitionRequestClient),这些对象都在NettyConnectionManager构造器中被初始化。

每个PartitionRequestClientFactory实例都依赖一个NettyClient。也就是说所有PartitionRequestClient底层都共用一个NettyClient。

Netty客户端和服务器对象的启动和停止都是由NettyConnectionManager统一控制的。NettyConnectionManager启动的时机是当TaskManager跟JobManager关联上之后调用NetworkEnvironment的associateWithTaskManagerAndJobManager方法时。而当TaskManager跟JobManager解除关联时停止。

NettyBufferPool

NettyClient和NettyServer在实例化Netty通信的核心对象时都需要配置各自的“字节缓冲分配器”用于为Netty读写数据分配内存单元。Netty自身提供了一个池化的字节缓冲分配器(PooledByteBufAllocator),但Flink又在此基础上进行了包装并提供了Netty缓冲池(NettyBufferPool)。此举的目的是严格控制所创建的分配器(Arena)的个数,转而依赖TaskManager的相关配置指定。

什么是Arena?当指定PooledByteBufAllocator来执行ByteBuf分配时,最终的内存分配工作被委托给类PoolArena。由于Netty通常用于高并发系统,所以各个线程进行内存分配时竞争不可避免,这可能会极大的影响内存分配的效率,为了缓解高并发时的线程竞争,Netty允许使用者创建多个分配器(Arena)来分离锁,提高内存分配效率。

NettyBufferPool在构造器内部以固定的参数实例化PooledByteBufAllocator并作为自己的内部分配器。具体做了哪些限制呢?首先,PooledByteBufAllocator本身既支持堆内存分配也支持堆外内存分配,NettyBufferPool将其限定为只在堆外内存上进行分配。其次, 显式指定了pageSize大小为8192,maxOrder值为11。这两个参数是什么意思呢?Netty中的内存池包含页(page)和块(chunk)两种分配单位,通过PooledByteBufAllocator构造器可以设置页大小(也即pageSize参数),该参数在PooledByteBufAllocator中的默认值为8192,而参数maxOder则用于计算块的大小。

计算公式为:chunkSize = pageSize << maxOrder;因此这里块大小为16MB。

另外,NettyBufferPool通过反射还拿到了PooledByteBufAllocator中的PoolArena分配器对象集合,但此举更多的是出于调试目的。并且显式关闭了对堆内存相关的操作方法。

NettyClient

NettyClient的主要职责是初始化Netty客户端的核心对象,并根据NettyProtocol配置用于客户端事件处理的ChannelPipeline。

NettyClient并不用于发起远程结果子分区请求,该工作将由PartitionRequestClient完成。

一个Netty引导客户端的创建步骤如下:

  • 创建Bootstrap对象用来引导启动客户端:
bootstrap = new Bootstrap();
  • 创建NioEventLoopGroup或EpollEventLoopGroup对象并设置到Bootstrap中,EventLoopGroup可以理解为是一个线程池,用来处理连接、接收数据、发送数据:
switch (config.getTransportType()) {
    case NIO:
        initNioBootstrap();
        break;

    case EPOLL:
        initEpollBootstrap();
        break;

    case AUTO:
        if (Epoll.isAvailable()) {
            initEpollBootstrap();
            LOG.info("Transport type 'auto': using EPOLL.");
        }
        else {
            initNioBootstrap();
            LOG.info("Transport type 'auto': using NIO.");
        }
}

Netty自版本4.0.16开始,对于Linux系统提供原生的套接字通信传输支持(也即,epoll机制,借助于JNI调用),这种传输机制拥有更高的性能。

  • 进行一系列配置,并设置ChannelHandler用来处理逻辑:
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel channel) throws Exception {
        channel.pipeline().addLast(protocol.getClientChannelHandlers());
    }
});

注意以上设置的是基于NettyPotocol获得的一个ChannelHandler数组组成的管道。

  • 调用Bootstrap.connect()来连接服务器:
return bootstrap.connect(serverSocketAddress);

以上就是一个Netty客户端从初始化到跟服务器建立连接的大致过程。但这里需要注意的是,一个TaskManager根本上只会存在一个NettyClient对象(对应的也只有一个Bootstrap实例)。但一个TaskManager中的子任务实例很有可能会跟多个不同的远程TaskManager通信,所以同一个Bootstrap实例可能会跟多个目标服务器建立连接,所以它是复用的,这一点不存在问题因为无论跟哪个目标服务器通信,Bootstrap的配置都是不变的。至于不同的RemoteChannel如何跟某个连接建立对应关系,这一点由PartitionRequestClientFactory来保证。

Netty自版本4.0.16开始,对于Linux系统提供原生的套接字通信传输支持(也即,epoll机制,借助于JNI调用),这种传输机制拥有更高的性能。

NettyServer

跟NettyClient一样,NettyServer也会初始化Netty服务端的核心对象,除此之外它会启动对特定端口的侦听并准备接收客户端发起的请求。下面是NettyServer的初始化与启动步骤:

  • 创建ServerBootstrap实例来引导绑定和启动服务器:
bootstrap = new ServerBootstrap();
  • 根据配置创建NioEventLoopGroup或EpollEventLoopGroup对象来处理事件,如接收新连接、接收数据、写数据等等:
switch (config.getTransportType()) {
    case NIO:
        initNioBootstrap();
        break;

    case EPOLL:
        initEpollBootstrap();
        break;

    case AUTO:
        if (Epoll.isAvailable()) {
            initEpollBootstrap();
            LOG.info("Transport type 'auto': using EPOLL.");
        }
        else {
            initNioBootstrap();
            LOG.info("Transport type 'auto': using NIO.");
        }
}
  • 指定InetSocketAddress,服务器监听此端口:
bootstrap.localAddress(config.getServerAddress(), config.getServerPort());
  • 进行各种参数配置,设置childHandler执行所有的连接请求:
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel channel) throws Exception {
        channel.pipeline().addLast(protocol.getServerChannelHandlers());
    }
});

注意以上设置的是基于NettyPotocol获得的一个ChannelHandler数组组成的管道。

  • 都设置完毕了,最后调用ServerBootstrap.bind()方法来绑定服务器:
bindFuture = bootstrap.bind().syncUninterruptibly();

原文发布时间为:2017-01-08
本文作者:vinoYang
本文来自云栖社区合作伙伴 CSDN博客,了解相关信息可以关注CSDN博客。
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
20天前
|
编解码 分布式计算 网络协议
Netty高性能网络框架(一)
Netty高性能网络框架(一)
|
2月前
|
Java 调度
Netty运行原理问题之ChannelHandler在Netty中扮演什么角色
Netty运行原理问题之ChannelHandler在Netty中扮演什么角色
|
2月前
|
编解码 网络协议 API
Netty运行原理问题之Netty的主次Reactor多线程模型工作的问题如何解决
Netty运行原理问题之Netty的主次Reactor多线程模型工作的问题如何解决
|
2月前
|
编解码 网络协议 开发者
Netty运行原理问题之NettyTCP的粘包和拆包的问题如何解决
Netty运行原理问题之NettyTCP的粘包和拆包的问题如何解决
|
2月前
|
调度
Netty运行原理问题之事件调度工作的问题如何解决
Netty运行原理问题之事件调度工作的问题如何解决
|
1天前
|
消息中间件 监控 Java
大数据-109 Flink 体系结构 运行架构 ResourceManager JobManager 组件关系与原理剖析
大数据-109 Flink 体系结构 运行架构 ResourceManager JobManager 组件关系与原理剖析
10 1
|
20天前
|
存储 机器人 Linux
Netty(二)-服务端网络编程常见网络IO模型讲解
Netty(二)-服务端网络编程常见网络IO模型讲解
|
2月前
|
数据采集 分布式计算 Kubernetes
Apache Flink 实践问题之ZooKeeper 网络瞬断时如何解决
Apache Flink 实践问题之ZooKeeper 网络瞬断时如何解决
52 4
|
2月前
|
存储 缓存 Java
实时计算 Flink版操作报错合集之怎么处理在运行作业时遇到报错::ClassCastException
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2月前
|
开发者
Netty运行原理问题之Netty高性能实现的问题如何解决
Netty运行原理问题之Netty高性能实现的问题如何解决