Mina2.0控制读取、写出流量代码（出自red5）

public class TrafficShapingFilter extends IoFilterAdapter {

?protected static Logger log = LoggerFactory.getLogger(TrafficShapingFilter.class);

?private final AttributeKey STATE = new AttributeKey(getClass(), "state");

?//定时器服务，用来创建定时任务的定时器服务类
?private final ScheduledExecutorService scheduledExecutor;

?//用来估算对象占据JVM内存的大小的类，计算出一个对象实际占用JVM内存的大小
?private final MessageSizeEstimator messageSizeEstimator;

?private volatile int maxReadThroughput;

?private volatile int maxWriteThroughput;

?private volatile int poolSize = 1;

?/**
? * 构造方法
? * @param maxReadThroughput 最大读取字节大小（单位：秒）
? * @param maxWriteThroughput 最大写出的字节大小（单位：秒）
? */
?public TrafficShapingFilter(int maxReadThroughput, int maxWriteThroughput) {
??this(null, null, maxReadThroughput, maxWriteThroughput);
?}

?public TrafficShapingFilter(ScheduledExecutorService scheduledExecutor, int maxReadThroughput,
???int maxWriteThroughput) {
??this(scheduledExecutor, null, maxReadThroughput, maxWriteThroughput);
?}

?public TrafficShapingFilter(ScheduledExecutorService scheduledExecutor, MessageSizeEstimator messageSizeEstimator,
???int maxReadThroughput, int maxWriteThroughput) {

??log.debug("ctor - executor: {} estimator: {} max read: {} max write: {}", new Object[] { scheduledExecutor,
????messageSizeEstimator, maxReadThroughput, maxWriteThroughput });

??if (scheduledExecutor == null) {
???//实例化一个定时器对象，线程池的默认数量是1.
???scheduledExecutor = new ScheduledThreadPoolExecutor(poolSize);
???//throw new NullPointerException("scheduledExecutor");
??}

??if (messageSizeEstimator == null) {
???//实例化默认的估算消息大小的类
???messageSizeEstimator = new DefaultMessageSizeEstimator() {
????@Override
????public int estimateSize(Object message) {
?????if (message instanceof IoBuffer) {
??????return ((IoBuffer) message).remaining();
?????}
?????return super.estimateSize(message);
????}
???};
??}

??this.scheduledExecutor = scheduledExecutor;
??this.messageSizeEstimator = messageSizeEstimator;
??//设置最大读取字节长度（单位：秒）
??setMaxReadThroughput(maxReadThroughput);
??//设置最大写出字节长度（单位：秒）
??setMaxWriteThroughput(maxWriteThroughput);
?}

?public ScheduledExecutorService getScheduledExecutor() {
??return scheduledExecutor;
?}

?public MessageSizeEstimator getMessageSizeEstimator() {
??return messageSizeEstimator;
?}

?public int getMaxReadThroughput() {
??return maxReadThroughput;
?}

?public void setMaxReadThroughput(int maxReadThroughput) {
??if (maxReadThroughput < 0) {
???maxReadThroughput = 0;
??}
??this.maxReadThroughput = maxReadThroughput;
?}

?public int getMaxWriteThroughput() {
??return maxWriteThroughput;
?}

?public void setMaxWriteThroughput(int maxWriteThroughput) {
??if (maxWriteThroughput < 0) {
???maxWriteThroughput = 0;
??}
??this.maxWriteThroughput = maxWriteThroughput;
?}

?public int getPoolSize() {
??return poolSize;
?}

?public void setPoolSize(int poolSize) {
??if (poolSize < 1) {
???poolSize = 1;
??}
??this.poolSize = poolSize;
?}

?@Override
?public void onPreAdd(IoFilterChain parent, String name, NextFilter nextFilter) throws Exception {
??if (parent.contains(this)) {
???throw new IllegalArgumentException(
?????"You can't add the same filter instance more than once.? Create another instance and add it.");
??}
??//给每一个session添加一个属性 STATE 属性，关联一个State对象。
??parent.getSession().setAttribute(STATE, new State());
??//调节会话sessiion读取buffer大小
??adjustReadBufferSize(parent.getSession());
?}

?@Override
?public void onPostRemove(IoFilterChain parent, String name, NextFilter nextFilter) throws Exception {
??//写完毕，关闭会话移除关联对象State
??parent.getSession().removeAttribute(STATE);
?}

?@Override
?public void messageReceived(NextFilter nextFilter, final IoSession session, Object message) throws Exception {

??int maxReadThroughput = this.maxReadThroughput;
??//process the request if our max is greater than zero
??if (maxReadThroughput > 0) {
???final State state = (State) session.getAttribute(STATE);
???long currentTime = System.currentTimeMillis();

???long suspendTime = 0;
???boolean firstRead = false;
???synchronized (state) {
????//估算当前已经读取字节数组总量
????state.readBytes += messageSizeEstimator.estimateSize(message);
???????????????? //如果读取回话挂起，suspendedRead=true
????if (!state.suspendedRead) {
?????if (state.readStartTime == 0) {//表示第一次读取会话的数据
??????firstRead = true;
??????//设置会话开始读取数据的时间
??????state.readStartTime = currentTime - 1000;
?????}
??????????????????? //估算当前平均每秒读取字节流量大小
?????long throughput = (state.readBytes * 1000 / (currentTime - state.readStartTime));
?????if (throughput >= maxReadThroughput) {//如果平均读取流量大于设置的数值，
??????//计算需要挂起的时间,((state.readBytes/maxReadThroughput)*1000)计算如果按照规定流量读取数据需要多少秒，然后再减去已经读取的时间差，计算出需要挂起的时间
??????suspendTime = Math.max(0, (state.readBytes * 1000 / maxReadThroughput)
????????- (firstRead ? 0 : currentTime - state.readStartTime));

??????state.readBytes = 0;
??????state.readStartTime = 0;
??????state.suspendedRead = suspendTime != 0;
??????????????????????? //调整会话状态，设置会话不在读取数据，利用定时器，挂起会话，
??????adjustReadBufferSize(session);
?????}
????}
???}

???if (suspendTime != 0) {
????session.suspendRead();
????scheduledExecutor.schedule(new Runnable() {
?????public void run() {
??????synchronized (state) {
???????state.suspendedRead = false;
??????}
??????session.resumeRead();
?????}
????}, suspendTime, TimeUnit.MILLISECONDS);
???}
??}

??nextFilter.messageReceived(session, message);

?}

?/**
? *
? * 调整session默认设置最大读取字节数组长度
? * @param session
? */
?private void adjustReadBufferSize(IoSession session) {
??int maxReadThroughput = this.maxReadThroughput;
??if (maxReadThroughput == 0) {//如果不限制读取流量返回
???return;
??}
??IoSessionConfig config = session.getConfig();
??if (config.getReadBufferSize() > maxReadThroughput) {
???config.setReadBufferSize(maxReadThroughput);
??}
??if (config.getMaxReadBufferSize() > maxReadThroughput) {
???config.setMaxReadBufferSize(maxReadThroughput);
??}
?}

?@Override
?public void messageSent(NextFilter nextFilter, final IoSession session, WriteRequest writeRequest) throws Exception {
????????? //得到当前系统设置最大写出字节
??int maxWriteThroughput = this.maxWriteThroughput;
??//process the request if our max is greater than zero
??if (maxWriteThroughput > 0) {
???final State state = (State) session.getAttribute(STATE);
???//得到系统时间
???long currentTime = System.currentTimeMillis();
??????????? //挂起时间长度0
???long suspendTime = 0;
???boolean firstWrite = false;
???synchronized (state) {
????state.writtenBytes += messageSizeEstimator.estimateSize(writeRequest.getMessage());
????if (!state.suspendedWrite) {
?????if (state.writeStartTime == 0) {
??????firstWrite = true;
??????//初始化写时间
??????state.writeStartTime = currentTime - 1000;
?????}
??????????????????? //计算平均写字节数组流量
?????long throughput = (state.writtenBytes * 1000 / (currentTime - state.writeStartTime));
?????if (throughput >= maxWriteThroughput) {//写流量超出系统设置，会话挂起操作
??????//计算会话需要挂起的时间
??????suspendTime = Math.max(0, state.writtenBytes * 1000 / maxWriteThroughput
????????- (firstWrite ? 0 : currentTime - state.writeStartTime));
??????state.writtenBytes = 0;
??????state.writeStartTime = 0;
??????state.suspendedWrite = suspendTime != 0;
?????}
????}
???}

???if (suspendTime != 0) {
????log.trace("Suspending write");
????//挂起会话
????session.suspendWrite();
????//定时器执行定时挂起操作
????scheduledExecutor.schedule(new Runnable() {
?????public void run() {
??????synchronized (state) {
???????state.suspendedWrite = false;
??????}
??????//挂起唤醒会话
??????session.resumeWrite();
??????log.trace("Resuming write");
?????}
????}, suspendTime, TimeUnit.MILLISECONDS);
???}
??}
??????? //串模式，执行下一个过滤器
??nextFilter.messageSent(session, writeRequest);

?}

?/**
? * 状态标志
? * @author Administrator
? *
? */
?private static class State {
??/**
?? * 开始读取数据的时间
?? */
??private long readStartTime;

??/**
?? * 开始写数据的时间
?? */
??private long writeStartTime;

??/**
?? * 读数据是否已经被挂起，true：挂起，false:未挂起
?? */
??private boolean suspendedRead;

??/**
?? * 是否写操作被挂起，true:被挂起，false:未被挂起
?? */
??private boolean suspendedWrite;

??/**
?? * 该会话总共被读取数据字节长度
?? */
??private long readBytes;

??/**
?? * 总共被写的数据的长度
?? */
??private long writtenBytes;
?}
}