- 浏览: 945726 次
文章分类
- 全部博客 (428)
- Hadoop (2)
- HBase (1)
- ELK (1)
- ActiveMQ (13)
- Kafka (5)
- Redis (14)
- Dubbo (1)
- Memcached (5)
- Netty (56)
- Mina (34)
- NIO (51)
- JUC (53)
- Spring (13)
- Mybatis (17)
- MySQL (21)
- JDBC (12)
- C3P0 (5)
- Tomcat (13)
- SLF4J-log4j (9)
- P6Spy (4)
- Quartz (12)
- Zabbix (7)
- JAVA (9)
- Linux (15)
- HTML (9)
- Lucene (0)
- JS (2)
- WebService (1)
- Maven (4)
- Oracle&MSSQL (14)
- iText (11)
- Development Tools (8)
- UTILS (4)
- LIFE (8)
最新评论
-
Donald_Draper:
Donald_Draper 写道刘落落cici 写道能给我发一 ...
DatagramChannelImpl 解析三(多播) -
Donald_Draper:
刘落落cici 写道能给我发一份这个类的源码吗Datagram ...
DatagramChannelImpl 解析三(多播) -
lyfyouyun:
请问楼主,执行消息发送的时候,报错:Transport sch ...
ActiveMQ连接工厂、连接详解 -
ezlhq:
关于 PollArrayWrapper 状态含义猜测:参考 S ...
WindowsSelectorImpl解析一(FdMap,PollArrayWrapper) -
flyfeifei66:
打算使用xmemcache作为memcache的客户端,由于x ...
Memcached分布式客户端(Xmemcached)
netty 事件执行器组和事件执行器定义及抽象实现:http://donald-draper.iteye.com/blog/2391257
netty 多线程事件执行器组:http://donald-draper.iteye.com/blog/2391270
netty 多线程事件循环组:http://donald-draper.iteye.com/blog/2391276
netty 抽象调度事件执行器:http://donald-draper.iteye.com/blog/2391379
netty 单线程事件执行器初始化:http://donald-draper.iteye.com/blog/2391895
netty 单线程事件执行器执行任务与graceful方式关闭:http://donald-draper.iteye.com/blog/2392051
引言:
上一篇文章看了单线程事件执行器的任务执行与执行器关闭,先来回顾一下:
单线程事件执行器,执行任务,首先判断任务是否为null,为空抛出空指针异常,否则,判断线程是否在当前事件循环中,在则添加任务到任务队列,否则开启当前单线程事件执行器,并添加任务到任务队列,如果此时事件执行器已关闭,并可以移除任务,则抛出拒绝执行器任务异常;如果需要启动事件执行器唤醒线程,则添加唤醒线程到任务队列。
添加,移除,poll任务操作,实际委托给任务队列,添加,移除hook线程操作委托给关闭hooks线程集合。
单线程事件执行器take任务,首先从调度任务队列peek头部调度任务,如果任务不为空,则获取调度任务延时时间,如果延时时间大于0,则从任务队列超时poll任务,否则从调度任务队列抓取调度任务,添加到任务队列,并从任务队列poll任务;如果调度任务为空,则从任务队列take一个任务,如果是唤醒任务,则忽略。
关闭单线程执行器,首先检查间隔、超时时间,时间单元参数,并且间隔时间要小于超时时间,如果已经关闭,则返回异步关闭任务结果,否则检查线程是否在当前事务循环中,如果是则更新状态为正在关闭,并计算计算关闭间隔和超时时间。
今天来看一下单线程事件循环SingleThreadEventLoop:
从上面来看,单线程事件循环SingleThreadEventLoop,继承了单线程事件执行器,实现了事件循环接口,内部一个事件循环任务队列,我们可以把单线程事件循环看为一个简单的事件执行器,单线程事件循环中多了一个通道注册的方法,实际注册工作委托给通道关联的UnSafe。
再来看一下Nio事件循环的定义:
nio事件循环实际为一个单线程事件循环,这样做的目的是,事件循环关联的通道注册到一个
选择器,可以复用循环事件,即保证通道的IO操作线程安全。
总结:
单线程事件循环SingleThreadEventLoop,继承了单线程事件执行器,实现了事件循环接口,
内部一个事件循环任务队列,我们可以把单线程事件循环看为一个简单的事件执行器,单线程事件循环中多了一个通道注册的方法,实际注册工作委托给通道关联的UnSafe。
netty 多线程事件执行器组:http://donald-draper.iteye.com/blog/2391270
netty 多线程事件循环组:http://donald-draper.iteye.com/blog/2391276
netty 抽象调度事件执行器:http://donald-draper.iteye.com/blog/2391379
netty 单线程事件执行器初始化:http://donald-draper.iteye.com/blog/2391895
netty 单线程事件执行器执行任务与graceful方式关闭:http://donald-draper.iteye.com/blog/2392051
引言:
上一篇文章看了单线程事件执行器的任务执行与执行器关闭,先来回顾一下:
单线程事件执行器,执行任务,首先判断任务是否为null,为空抛出空指针异常,否则,判断线程是否在当前事件循环中,在则添加任务到任务队列,否则开启当前单线程事件执行器,并添加任务到任务队列,如果此时事件执行器已关闭,并可以移除任务,则抛出拒绝执行器任务异常;如果需要启动事件执行器唤醒线程,则添加唤醒线程到任务队列。
添加,移除,poll任务操作,实际委托给任务队列,添加,移除hook线程操作委托给关闭hooks线程集合。
单线程事件执行器take任务,首先从调度任务队列peek头部调度任务,如果任务不为空,则获取调度任务延时时间,如果延时时间大于0,则从任务队列超时poll任务,否则从调度任务队列抓取调度任务,添加到任务队列,并从任务队列poll任务;如果调度任务为空,则从任务队列take一个任务,如果是唤醒任务,则忽略。
关闭单线程执行器,首先检查间隔、超时时间,时间单元参数,并且间隔时间要小于超时时间,如果已经关闭,则返回异步关闭任务结果,否则检查线程是否在当前事务循环中,如果是则更新状态为正在关闭,并计算计算关闭间隔和超时时间。
今天来看一下单线程事件循环SingleThreadEventLoop:
package io.netty.channel; import io.netty.util.concurrent.RejectedExecutionHandler; import io.netty.util.concurrent.RejectedExecutionHandlers; import io.netty.util.concurrent.SingleThreadEventExecutor; import io.netty.util.internal.ObjectUtil; import io.netty.util.internal.SystemPropertyUtil; import io.netty.util.internal.UnstableApi; import java.util.Queue; import java.util.concurrent.Executor; import java.util.concurrent.ThreadFactory; /** * Abstract base class for {@link EventLoop}s that execute all its submitted tasks in a single thread. 单线程事件循环在单个线程中执行所有提交的任务 * */ public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop { protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16, SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE)); //当前循环任务队列 private final Queue<Runnable> tailTasks; //下面几个构造方法与单线程事件执行器基本相同,就不说了 protected SingleThreadEventLoop(EventLoopGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) { this(parent, threadFactory, addTaskWakesUp, DEFAULT_MAX_PENDING_TASKS, RejectedExecutionHandlers.reject()); } protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor, boolean addTaskWakesUp) { this(parent, executor, addTaskWakesUp, DEFAULT_MAX_PENDING_TASKS, RejectedExecutionHandlers.reject()); } protected SingleThreadEventLoop(EventLoopGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedExecutionHandler) { super(parent, threadFactory, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler); tailTasks = newTaskQueue(maxPendingTasks); } protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor, boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedExecutionHandler) { super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler); tailTasks = newTaskQueue(maxPendingTasks); } //获取所属事件循环组 @Override public EventLoopGroup parent() { return (EventLoopGroup) super.parent(); } //获取当前事件循环 @Override public EventLoop next() { return (EventLoop) super.next(); } //注册通道 @Override public ChannelFuture register(Channel channel) { return register(new DefaultChannelPromise(channel, this)); } //注册通道 @Override public ChannelFuture register(final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, "promise"); //委托给通道关联的UnSafe,这个具体我们在后面再说 promise.channel().unsafe().register(this, promise); return promise; } //此注册方法已丢弃 @Deprecated @Override public ChannelFuture register(final Channel channel, final ChannelPromise promise) { if (channel == null) { throw new NullPointerException("channel"); } if (promise == null) { throw new NullPointerException("promise"); } channel.unsafe().register(this, promise); return promise; } /** * Adds a task to be run once at the end of next (or current) {@code eventloop} iteration. *在事件循环迭代后,运行指定任务 * @param task to be added. */ @UnstableApi public final void executeAfterEventLoopIteration(Runnable task) { ObjectUtil.checkNotNull(task, "task"); if (isShutdown()) { //事件循环关闭,则抛出拒绝执行异常 reject(); } //如果添加任务到任务队列失败,则拒绝执行任务 if (!tailTasks.offer(task)) { reject(task); } //如果需要为任务,启动唤醒线程,则添加唤醒线程到任务队列 if (wakesUpForTask(task)) { wakeup(inEventLoop()); } } /** * Removes a task that was added previously via {@link #executeAfterEventLoopIteration(Runnable)}. *在在事件循环迭代后,移除指定任务 * @param task to be removed. * * @return {@code true} if the task was removed as a result of this call. */ @UnstableApi final boolean removeAfterEventLoopIterationTask(Runnable task) { return tailTasks.remove(ObjectUtil.checkNotNull(task, "task")); } //是否启动唤醒任务线程 @Override protected boolean wakesUpForTask(Runnable task) { return !(task instanceof NonWakeupRunnable); } //在运行所有任务结束后,执行tailTasks任务队列中的任务 @Override protected void afterRunningAllTasks() { runAllTasksFrom(tailTasks); } //判断tailTasks任务队列中是否有任务 @Override protected boolean hasTasks() { return super.hasTasks() || !tailTasks.isEmpty(); } //获取任务队列中的任务数 @Override public int pendingTasks() { return super.pendingTasks() + tailTasks.size(); } /** * Marker interface for {@link Runnable} that will not trigger an {@link #wakeup(boolean)} in all cases. 标记线程不会触发唤醒线程 */ interface NonWakeupRunnable extends Runnable { } }
从上面来看,单线程事件循环SingleThreadEventLoop,继承了单线程事件执行器,实现了事件循环接口,内部一个事件循环任务队列,我们可以把单线程事件循环看为一个简单的事件执行器,单线程事件循环中多了一个通道注册的方法,实际注册工作委托给通道关联的UnSafe。
再来看一下Nio事件循环的定义:
/** * {@link SingleThreadEventLoop} implementation which register the {@link Channel}'s to a * {@link Selector} and so does the multi-plexing of these in the event loop. * */ public final class NioEventLoop extends SingleThreadEventLoop {
nio事件循环实际为一个单线程事件循环,这样做的目的是,事件循环关联的通道注册到一个
选择器,可以复用循环事件,即保证通道的IO操作线程安全。
总结:
单线程事件循环SingleThreadEventLoop,继承了单线程事件执行器,实现了事件循环接口,
内部一个事件循环任务队列,我们可以把单线程事件循环看为一个简单的事件执行器,单线程事件循环中多了一个通道注册的方法,实际注册工作委托给通道关联的UnSafe。
发表评论
-
netty NioSocketChannel解析
2017-09-29 12:50 1221netty 抽象BootStrap定义:http://dona ... -
netty Pooled字节buf分配器
2017-09-28 13:00 1956netty 字节buf定义:http://donald-dra ... -
netty Unpooled字节buf分配器
2017-09-26 22:01 2342netty 字节buf定义:http://donald-dra ... -
netty 抽象字节buf分配器
2017-09-26 08:43 1247netty 字节buf定义:http:// ... -
netty 复合buf概念
2017-09-25 22:31 1241netty 字节buf定义:http://donald-dra ... -
netty 抽象字节buf引用计数器
2017-09-22 12:48 1498netty 字节buf定义:http://donald-dra ... -
netty 抽象字节buf解析
2017-09-22 09:00 1743netty 通道接口定义:http://donald-drap ... -
netty 资源泄漏探测器
2017-09-21 09:37 1314netty 通道接口定义:http://donald-drap ... -
netty 字节buf定义
2017-09-20 08:31 2741netty 通道接口定义:http://donald-drap ... -
netty 默认通道配置后续
2017-09-18 08:36 2080netty 通道接口定义:http://donald-drap ... -
netty 默认通道配置初始化
2017-09-17 22:51 1928netty 通道接口定义:http://donald-drap ... -
netty 通道配置接口定义
2017-09-17 14:51 997netty 通道接口定义:http://donald-drap ... -
netty NioServerSocketChannel解析
2017-09-16 13:01 1798netty ServerBootStrap解析:http:// ... -
netty 抽象nio消息通道
2017-09-15 15:30 1145netty 通道接口定义:http:/ ... -
netty 抽象nio字节通道
2017-09-14 22:39 1133netty 通道接口定义:http:/ ... -
netty 抽象nio通道解析
2017-09-14 17:23 883netty 通道接口定义:http://donald-drap ... -
netty 抽象通道后续
2017-09-13 22:40 1220netty Inboudn/Outbound通道Inv ... -
netty 通道Outbound缓冲区
2017-09-13 14:31 2115netty 通道接口定义:http:/ ... -
netty 抽象Unsafe定义
2017-09-12 21:24 983netty 通道接口定义:http:/ ... -
netty 抽象通道初始化
2017-09-11 12:56 1772netty 管道线定义-ChannelPipeline:htt ...
相关推荐
Netty多线程案例集锦
Netty多线程并发编程
Netty服务器线程模型概览_线程模型
netty5深入源码解析多线程编程,netty5高性能并发编程的首选
使用Netty4实现多线程的消息分发,这是一个基于netty4做的一个异步通信模型。
netty中的多线程并发应用,
这是一本详细介绍了netty线程模型的书籍,包括 nio核心
netty的简介,netty是一种socket技术,实现长连接,此技术更加成熟1
Reactor单线程模型,指的是所有的IO操作都在同一个NIO线程上面完成,NIO线程的职责如下: 1)作为NIO服务端,接收客户端的TCP连接; 2)作为NIO客户端,向服务端发起TCP连接; 3)读取通信对端的请求或者应答消息;...
深入Hotspot源码与Linux内核理解NIO与Netty线程模型
Netty是一个提供异步事件驱动的网络应用框架,用以快速开发高性能、高可靠性的网络服务器和客户端程序。 换句话说,Netty是一个NIO框架,使用它可以简单快速地开发网络应用程序,比如客户端和服务端的协议。Netty...
Netty 通讯框架 多线程 源代码 刚写的一个通讯框架,可以启动多个client,并设置client数量,可以用于测试Netty框架支持多少用户数量
Netty核心精讲之Reactor线程模型源码分析 Netty核心精讲之Reactor线程模型源码分析
05.Netty线程模型.rar
Netty线程模型和EventLoop线程模型概述EventLoop事件循环任务调度JDK任务调度EventLoop任务调度线程管理线程分配非阻塞传输阻塞传输N
Netty源码解读之线程
Netty 消息接收和线程处理模型
Netty基础,用于学习Netty,参考黑马程序员的netty教程
再根据本人实际学习体验总结而成。本部分内容可能不那么全面,但是我尽量挑选Netty中我认为比较重要的部分做讲解。
多线程并发编程在Netty中的应用分析