lettuce和hbase中对netty的使用你都了解吗？
之前的两篇文章中咱们有详细地聊过关于lettuce的pipeline以及spring-data-redis对其封装的细节。本篇紧接着上面篇以connectionPoolingProvider为入口，对lettuce基于netty处理IO事件的线程池进行进一步地分析。
lettuce EventLoopGroup初始化
在使用lettuce作为redis连接池时，在上一节中我们知道，lettuce中维护连接有两种使用连接池的方式，目前一种已经废弃，另一种大家正在使用的版本是apache commons pool。咱们来回顾下。
从连接池获取连接的入口
org.springframework.data.redis.connection.lettuce.LettucePoolingConnectionProvider#getConnection：
@Override
public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
GenericObjectPool<StatefulConnection<?, ?>> pool = pools.computeIfAbsent(connectionType, poolType -> {
// 不存在就创建连接池
return ConnectionPoolSupport.createGenericObjectPool(() -> connectionProvider.getConnection(connectionType),
poolConfig, false);
});
try {
// 从连接池中获取
StatefulConnection<?, ?> connection = pool.borrowObject();
// 放入map中
poolRef.put(connection, pool);
return connectionType.cast(connection);
} catch (Exception e) {
throw new PoolException("Could not get a resource from the pool", e);
}
}上面的代码中获取连接的核心是connectionProvider.getConnection(connectionType)，我们先来看一下connectionProvider的分类列表如下：
LettucePoolingConnectionProvider只是一个代理，在它的内部维护着具体的创建连接的provider。在我的场景下的这个provider是StandaloneConnectionProvider。
我们来看一下org.springframework.data.redis.connection.lettuce.StandaloneConnectionProvider#getConnection(java.lang.Class)方法：
@Override
public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
------------省略部分--------------
if (StatefulConnection.class.isAssignableFrom(connectionType)) {
return connectionType.cast(readFrom.map(it -> this.masterReplicaConnection(redisURISupplier.get(), it))
.orElseGet(() -> client.connect(codec)));
}
-------------------------
}在这里是使用RedisClient获取连接的方法，接着一起来看下io.lettuce.core.RedisClient#connect(io.lettuce.core.codec.RedisCodec<K,V>)方法：
public <K, V> StatefulRedisConnection<K, V> connect(RedisCodec<K, V> codec) {
checkForRedisURI();
return getConnection(connectStandaloneAsync(codec, this.redisURI, timeout));
}在这里connectStandaloneAsync方法返回的是一个ConnectionFuture<StatefulRedisConnection<K, V>>对象，是一个维户着我们的redis连接的Future。来看下io.lettuce.core.RedisClient#connectStandaloneAsync方法：
private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandaloneAsync(RedisCodec<K, V> codec,
RedisURI redisURI, Duration timeout) {
-------------------
// 创建DefaultEndpoint对象
DefaultEndpoint endpoint = new DefaultEndpoint(clientOptions, clientResources);
RedisChannelWriter writer = endpoint;
if (CommandExpiryWriter.isSupported(clientOptions)) {
writer = new CommandExpiryWriter(writer, clientOptions, clientResources);
}
// 创建连接对象
StatefulRedisConnectionImpl<K, V> connection = newStatefulRedisConnection(writer, codec, timeout);
// 进行连接
ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStatefulAsync(connection, codec, endpoint, redisURI,
() -> new CommandHandler(clientOptions, clientResources, endpoint));
------------------------------------------
return future;
}这里我们来分析下真正去建立连接的过程，主要逻辑在io.lettuce.core.RedisClient#connectStatefulAsync方法中，代码如下：
private <K, V, S> ConnectionFuture<S> connectStatefulAsync(StatefulRedisConnectionImpl<K, V> connection,
RedisCodec<K, V> codec, Endpoint endpoint,
RedisURI redisURI, Supplier<CommandHandler> commandHandlerSupplier) {
ConnectionBuilder connectionBuilder;
if (redisURI.isSsl()) {
-------------省略部分代码----------------
} else {
// 创建connection builder对象
connectionBuilder = ConnectionBuilder.connectionBuilder();
}
connectionBuilder.connection(connection);
connectionBuilder.clientOptions(clientOptions);
connectionBuilder.clientResources(clientResources);
connectionBuilder.commandHandler(commandHandlerSupplier).endpoint(endpoint);
// 在这里会创建netty client Bootstrap对象
connectionBuilder(getSocketAddressSupplier(redisURI), connectionBuilder, redisURI);
// 在这里根据channel type来初始化eventLoopGroup对象
channelType(connectionBuilder, redisURI);
----------------------省略部分代码--------------------
return sync.thenApply(channelHandler -> (S) connection);
}这里主要需要分析个方法：connectionBuilder方法和channelType方法。
1.io.lettuce.core.AbstractRedisClient#connectionBuilder方法：
protected void connectionBuilder(Mono<SocketAddress> socketAddressSupplier, ConnectionBuilder connectionBuilder,
RedisURI redisURI) {
// 创建bootstrap对象
Bootstrap redisBootstrap = new Bootstrap();
// 设置高水位线
redisBootstrap.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
// 设置低水平线
redisBootstrap.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
// buf的allocator
redisBootstrap.option(ChannelOption.ALLOCATOR, BUF_ALLOCATOR);
SocketOptions socketOptions = getOptions().getSocketOptions();
// 设置bootstrap连接超时时间
redisBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
Math.toIntExact(socketOptions.getConnectTimeout().toMillis()));
if (LettuceStrings.isEmpty(redisURI.getSocket())) {
redisBootstrap.option(ChannelOption.SO_KEEPALIVE, socketOptions.isKeepAlive());
redisBootstrap.option(ChannelOption.TCP_NODELAY, socketOptions.isTcpNoDelay());
}
// 设置超时
connectionBuilder.timeout(redisURI.getTimeout());
// 设置密码
connectionBuilder.password(redisURI.getPassword());
connectionBuilder.bootstrap(redisBootstrap);
connectionBuilder.channelGroup(channels).connectionEvents(connectionEvents).timer(timer);
connectionBuilder.socketAddressSupplier(socketAddressSupplier);
}
这里主要是初始化 netty client的相关对象，并对其中一些参数进行设置。可以看到整个connectionBuilder对象中维护着连接所需要的所有信息。
1.io.lettuce.core.AbstractRedisClient#channelType方法
protected void channelType(ConnectionBuilder connectionBuilder, ConnectionPoint connectionPoint) {
LettuceAssert.notNull(connectionPoint, "ConnectionPoint must not be null");
connectionBuilder.bootstrap().group(getEventLoopGroup(connectionPoint));
if (connectionPoint.getSocket() != null) {
NativeTransports.assertAvailable();
connectionBuilder.bootstrap().channel(NativeTransports.domainSocketChannelClass());
} else {
connectionBuilder.bootstrap().channel(Transports.socketChannelClass());
}
}这个方法的目的也很简单，主要是设置执行netty 网络IO操作的线程池 eventLoopGroup。在真正开始分析eventLoopGroup之前，先来看一下io.lettuce.core.Transports#eventLoopGroupClass方法：
static Class<? extends EventLoopGroup> eventLoopGroupClass() {
if (NativeTransports.isSocketSupported()) {
return NativeTransports.eventLoopGroupClass();
}
return NioEventLoopGroup.class;
}这个方法的主要目的是确定使用哪种类型的NioEventLoopGroup。在io.lettuce.core.Transports.NativeTransports#eventLoopGroupClass方法内部：
static Class<? extends EventLoopGroup> eventLoopGroupClass() {
if (KqueueProvider.isAvailable()) {
return KqueueProvider.eventLoopGroupClass();
}
return EpollProvider.eventLoopGroupClass();
}这个方法会先判断kqueue与epoll provider的可用性，然后来决定eventLoopGroup的class类型是io.netty.channel.kqueue.KQueueEventLoopGroup还是io.netty.channel.epoll.EpollEventLoopGroup。具体判断的逻辑在EpollProvider和KqueueProvider的静态代码块，有兴趣的可以自己去分析一下，这里简单地看下kqueue的:
如果既没有epoll也没有kqueue，那么会使用NioEventLoopGroup。
获取EventLoopGroup
此时再回过头来看下io.lettuce.core.AbstractRedisClient#getEventLoopGroup方法代码如下：
private synchronized EventLoopGroup getEventLoopGroup(ConnectionPoint connectionPoint) {
if (connectionPoint.getSocket() == null && !eventLoopGroups.containsKey(Transports.eventLoopGroupClass())) {
// 先进行对应eventLoopGroupClass的初始化
eventLoopGroups.put(Transports.eventLoopGroupClass(),clientResources.eventLoopGroupProvider().allocate(Transports.eventLoopGroupClass()));
}
if (connectionPoint.getSocket() != null) {
NativeTransports.assertAvailable();
Class<? extends EventLoopGroup> eventLoopGroupClass = NativeTransports.eventLoopGroupClass();
if (!eventLoopGroups.containsKey(NativeTransports.eventLoopGroupClass())) {
eventLoopGroups
.put(eventLoopGroupClass, clientResources.eventLoopGroupProvider().allocate(eventLoopGroupClass));
}
}
if (connectionPoint.getSocket() == null) {
// 返回
return eventLoopGroups.get(Transports.eventLoopGroupClass());
}
if (connectionPoint.getSocket() != null) {
NativeTransports.assertAvailable();
return eventLoopGroups.get(NativeTransports.eventLoopGroupClass());
}
throw new IllegalStateException("This should not have happened in a binary decision. Please file a bug.");
}这里我们主要关注clientResources.eventLoopGroupProvider().allocate(Transports.eventLoopGroupClass()）方法，其中clientResources.eventLoopGroupProvider()方法会从clientResources中获取到eventLoopGroupProvider，然后使用这个provider去申请对应class类型的eventLoopGroup。
在DefaultClientResources中的provider类型是DefaultEventLoopGroupProvider，至于为什么这样，下文会有分析。我们来看下io.lettuce.core.resource.DefaultEventLoopGroupProvider#allocate方法：
@Override
public <T extends EventLoopGroup> T allocate(Class<T> type) {
synchronized (this) {
logger.debug("Allocating executor {}", type.getName());
// 将引用放入到外部容器中去，方便计数统计
return addReference(getOrCreate(type));
}
}
io.lettuce.core.resource.DefaultEventLoopGroupProvider#getOrCreate:
private <T extends EventLoopGroup> T getOrCreate(Class<T> type) {
---------------省略部分代码------------------
if (!eventLoopGroups.containsKey(type)) {
// 如果不存在，则创建eventLoopGroup
eventLoopGroups.put(type, createEventLoopGroup(type, numberOfThreads));
}
return (T) eventLoopGroups.get(type);
}
io.lettuce.core.resource.DefaultEventLoopGroupProvider#createEventLoopGroup:
public static <T extends EventExecutorGroup> EventExecutorGroup createEventLoopGroup(Class<T> type, int numberOfThreads) {
logger.debug("Creating executor {}", type.getName());
if (DefaultEventExecutorGroup.class.equals(type)) {
return new DefaultEventExecutorGroup(numberOfThreads, new DefaultThreadFactory("lettuce-eventExecutorLoop", true));
}
if (NioEventLoopGroup.class.equals(type)) {
return new NioEventLoopGroup(numberOfThreads, new DefaultThreadFactory("lettuce-nioEventLoop", true));
}
if (EpollProvider.isAvailable() && EpollProvider.isEventLoopGroup(type)) {
return EpollProvider.newEventLoopGroup(numberOfThreads, new DefaultThreadFactory("lettuce-epollEventLoop", true));
}
if (KqueueProvider.isAvailable() && KqueueProvider.isEventLoopGroup(type)) {
return KqueueProvider.newEventLoopGroup(numberOfThreads, new DefaultThreadFactory("lettuce-kqueueEventLoop", true));
}
throw new IllegalArgumentException(String.format("Type %s not supported", type.getName()));
}这里的getEventLoopGroup方法会生成真正供netty使用的EventLoopGroup,可能是nio、epoll、kqueue中的一种。先判断下当前环境是否支持epoll和kequeue，如果支持则会生成对应的eventLoopGroup，默认使用的是NioEventLoopGroup。没错，这个eventLoopGroup就是netty client进行io操作的eventLoopGroup。
DefaultClientResources 的 DefaultEventExecutorGroup
这里先来看一下clientResources的初始化，org.springframework.boot.autoconfigure.data.redis.LettuceConnectionConfiguration#lettuceClientResources代码:
@Bean(destroyMethod = "shutdown")
@ConditionalOnMissingBean(ClientResources.class)
public DefaultClientResources lettuceClientResources() {
return DefaultClientResources.create();
}
io.lettuce.core.resource.DefaultClientResources#create:
public static DefaultClientResources create() {
return builder().build();
}
io.lettuce.core.resource.DefaultClientResources.Builder#build:
@Override
public DefaultClientResources build() {
return new DefaultClientResources(this);
}
protected DefaultClientResources(Builder builder) {
if (builder.eventLoopGroupProvider == null) {
int ioThreadPoolSize = builder.ioThreadPoolSize;
----------省略部分代码---------
// 创建DefaultEventLoopGroupProvider
this.eventLoopGroupProvider = new DefaultEventLoopGroupProvider(ioThreadPoolSize);
} else {
this.sharedEventLoopGroupProvider = builder.sharedEventLoopGroupProvider;
this.eventLoopGroupProvider = builder.eventLoopGroupProvider;
}
if (builder.eventExecutorGroup == null) {
------------------------------
// 创建eventExecutorGroup
eventExecutorGroup = DefaultEventLoopGroupProvider.createEventLoopGroup(DefaultEventExecutorGroup.class,
computationThreadPoolSize);
sharedEventExecutor = false;
} else {
sharedEventExecutor = builder.sharedEventExecutor;
eventExecutorGroup = builder.eventExecutorGroup;
}
if (builder.timer == null) {
// 创建时间轮timer
timer = new HashedWheelTimer(new DefaultThreadFactory("lettuce-timer"));
sharedTimer = false;
} else {
timer = builder.timer;
sharedTimer = builder.sharedTimer;
}
if (builder.eventBus == null) {
// 创建事件驱动服务总线
eventBus = new DefaultEventBus(Schedulers.fromExecutor(eventExecutorGroup));
} else {
eventBus = builder.eventBus;
}
-------------------------------------这里本文主要关注的重点是创建eventExecutorGroup的方法，eventExecutorGroup是EventExecutorGroup类型的，它的主要作用是作为netty channelGroup的EventExecutor。关于EventExecutor参考
•https://www.cnblogs.com/lighten/p/8967630.html•https://juejin.im/post/5de7a0dcf265da33bf1ff42c
io.lettuce.core.resource.DefaultEventLoopGroupProvider#createEventLoopGroup方法代码如下：
public static <T extends EventExecutorGroup> EventExecutorGroup createEventLoopGroup(Class<T> type, int numberOfThreads) {
logger.debug("Creating executor {}", type.getName());
if (DefaultEventExecutorGroup.class.equals(type)) {
// 创建默认的eventExecutorGroup
return new DefaultEventExecutorGroup(numberOfThreads, new DefaultThreadFactory("lettuce-eventExecutorLoop", true));
}
if (NioEventLoopGroup.class.equals(type)) {
// 如果是NioEventLoopGroup类型的会创建NioEventLoopGroup
return new NioEventLoopGroup(numberOfThreads, new DefaultThreadFactory("lettuce-nioEventLoop", true));
}
if (EpollProvider.isAvailable() && EpollProvider.isEventLoopGroup(type)) {
// 创建epoll类型的eventLoop
return EpollProvider.newEventLoopGroup(numberOfThreads, new DefaultThreadFactory("lettuce-epollEventLoop", true));
}
if (KqueueProvider.isAvailable() && KqueueProvider.isEventLoopGroup(type)) {
// 创建kqueue类型的eventLoop
return KqueueProvider.newEventLoopGroup(numberOfThreads, new DefaultThreadFactory("lettuce-kqueueEventLoop", true));
}
throw new IllegalArgumentException(String.format("Type %s not supported", type.getName()));
}这个方法的主要作用是根据类型创建相应类型的EventLoopGroup，对于DefaultClientResources来说，它创建的是DefaultEventExecutorGroup，属于EventExecutorGroup类型的，它的主要作用是处理一些回调事件，异步处理，超时处理等，但是它不是用于处理netty io事件的那个eventLoopGroup。
说在后面的话
hbase client中底层rpc也是使用的也是Netty，它的org.apache.hadoop.hbase.ipc.NettyRpcConnection#connect方法部分截图如下：
这里也会涉及到一个rpcClient.group的初始化过程，它和lettuce中对netty的使用有什么区别呢？感兴趣的可以去分析一下。
- DeepLearning.ai学习笔记（一）神经网络和深度学习--Week4深层神经网络
- 博客园自定义皮肤扁平化设计
- DeepLearning.ai学习笔记（四）卷积神经网络 -- week3 目标检测
- DeepLearning.ai学习笔记（四）卷积神经网络 -- week2深度卷积神经网络 实例探究
- DeepLearning.ai学习笔记（四）卷积神经网络 -- week1 卷积神经网络基础知识介绍
- Udacity并行计算课程笔记-The GPU Hardware and Parallel Communication Patterns
- DeepLearning.ai学习笔记（三）结构化机器学习项目--week2机器学习策略(2)
- DeepLearning.ai学习笔记（三）结构化机器学习项目--week1 机器学习策略
- coursera 视频总是缓冲或者无法观看的解决办法
- DeepLearning.ai学习笔记（二）改善深层神经网络：超参数调试、正则化以及优化--week3 超参数调试、Batch正则化和程序框架
- DeepLearning.ai学习笔记（一）神经网络和深度学习--Week3浅层神经网络
- Andrew Ng机器学习课程笔记--week3（逻辑回归&正则化参数）
- Andrew Ng机器学习课程笔记--week1（机器学习介绍及线性回归）
- C++学习笔记之模板篇
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- maven 打包过滤二进制压缩与静态文件失效的事情
- The container name “/xx“ is already in use by container “xx“. You have to remove (or rename) that
- ClickHouse的可视化工具Tabix
- 跳表
- 浅谈AQS
- linux进程调度
- 696. 计数二进制子串 Krains 2020-08-10 09:42:23 字符串
- 给你的热图挑选一个合适的渐变色
- 网易云解锁灰色音乐
- AndroidStdio1_4
- 130. 被围绕的区域 Krains 2020-08-11 10:50:01 并查集DFS
- scRNA-seq Clustering(二)
- Apache-Hive 使用MySQL存储Hive的元数据
- 使用TensorFlow创建能够图像重建的自编码器模型
- Qt音视频开发9-ffmpeg录像存储