5、看！源码之netty中future之ChannelGroup
2019-01-29 本文已影响0人
starskye
ChannelGroupFuture接口实现
直接已经介绍过ChannelGroupFuture接口的定义这里将不会再重复讲述，第一个类是VoidChannelGroupFuture这个类并没有什么好说的，和之前的VoidChannelPromise一样是无效的类型，此类中没有什么特别可讲因为大多都没有逻辑进行执行，所以这里将不会讲述此类，直接开始讲述DefaultChannelGroupFuture类。
在讲述在之前需要讲述下它内部引用的一些定义。
//此接口是对Channel对象做的分组，分了组每一组都是一个ChannelGroup而他可以进行批量管理
public interface ChannelGroup extends Set<Channel>, Comparable<ChannelGroup> {
//返回当前分组的名字
String name();
//根据ChannelId获取对于的Channel,Channel接口定义的时候就要一个id方法是用来返回当前的Channel的id
Channel find(ChannelId id);
//给当前组内的Channel群发消息，如果消息是ByteBuf类型将会不会出现发送内容错误的情况，因为他会将内容进行一份保存每次返回的都是最原始的对象
//并且此方法是异步的，因为Channel中的write是异步的。
ChannelGroupFuture write(Object message);
//是上方方法的重载，增加了匹配，用于匹配channel，这样就可以在组内匹配指定channel进行发送消息
ChannelGroupFuture write(Object message, ChannelMatcher matcher);
//上方方法的重载，添加了是否返回无效的应答，此应答就是VoidChannelGroupFuture，如果是false那么将返回DefaultChannelGroupFuture
ChannelGroupFuture write(Object message, ChannelMatcher matcher, boolean voidPromise);
//刷新组内全部的channel管道，说白了就是循环channel集合然后调用每个channel的flush方法
ChannelGroup flush();
//上方方法的重载刷新匹配到的管道
ChannelGroup flush(ChannelMatcher matcher);
//结合了write与flush方法
ChannelGroupFuture writeAndFlush(Object message);
//此方法已经废弃了内部实现调用的是writeAndFlush方法
@Deprecated
ChannelGroupFuture flushAndWrite(Object message);
//结合flush(ChannelMatcher matcher)与write(Object message, ChannelMatcher matcher)方法，可以过滤管道
ChannelGroupFuture writeAndFlush(Object message, ChannelMatcher matcher);
//结合flush(ChannelMatcher matcher);与write(Object message, ChannelMatcher matcher, boolean voidPromise);方法。
ChannelGroupFuture writeAndFlush(Object message, ChannelMatcher matcher, boolean voidPromise);
//内部调用writeAndFlush(Object message, ChannelMatcher matcher)方法，此方法已废弃
@Deprecated
ChannelGroupFuture flushAndWrite(Object message, ChannelMatcher matcher);
//操作组内所有的channel断开连接
ChannelGroupFuture disconnect();
//上方方法的重构只是添加了筛选channel的功能
ChannelGroupFuture disconnect(ChannelMatcher matcher);
//关闭channel管道
ChannelGroupFuture close();
//重构上方方法添加了管道的过滤
ChannelGroupFuture close(ChannelMatcher matcher);
//用于注销事件使用EventLoop中的
@Deprecated
ChannelGroupFuture deregister();
//重载上方代码加入管道筛选已经废弃不建议使用
@Deprecated
ChannelGroupFuture deregister(ChannelMatcher matcher);
//获取管道关闭时的Future，可以使用它监听关闭
ChannelGroupFuture newCloseFuture();
//重载上方方法获取指定管道的关闭Future
ChannelGroupFuture newCloseFuture(ChannelMatcher matcher);
}
//默认管道组的操作管理，此类是提供给ChannelGroup接口使用的，因为ChannelGroup是批量管理自然也需要有个能够批量管理的Future而此类就是这个Future
final class DefaultChannelGroupFuture extends DefaultPromise<Void> implements ChannelGroupFuture {
//管理的操作集合是哪个ChannelGroup的，此变量暂无意义，因为本章并未使用
private final ChannelGroup group;
//当前组中的结果管理的所有管道和管道处理
private final Map<Channel, ChannelFuture> futures;
//本组任务执行成功个数和失败个数
private int successCount;
private int failureCount;
//默认监听器的实现，用于记录执行结果，在futures中的每个任务执行完成都会操作此监听器用于记录操作结果，比如成功数和失败数与失败结果的封装
private final ChannelFutureListener childListener = new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
//获取当前任务的执行结果
boolean success = future.isSuccess();
//声明是否是最后一个处理结果
boolean callSetDone;
//使用了DefaultChannelGroupFuture为锁是为了确保上方的计数变量的准确性
synchronized (DefaultChannelGroupFuture.this) {
//如果是成功则成功计数加一否则失败计数加一
if (success) {
successCount ++;
} else {
failureCount ++;
}
//如果成功数和失败数的和等于总处理数则返回true否则返回false
callSetDone = successCount + failureCount == futures.size();
//断言他的结果必然是小于等于处理任务，如果出现大于那么逻辑肯定是有问题的需要抛出断言异常，当然前提是运行时开启了断言
assert successCount + failureCount <= futures.size();
}
//如果当前处理是最后一个结果
if (callSetDone) {
//判断当前失败的个数是否大于0
if (failureCount > 0) {
//创建一个集合用于存储执行结果失败的管道与异常信息，设置个数为失败个数
List<Map.Entry<Channel, Throwable>> failed =
new ArrayList<Map.Entry<Channel, Throwable>>(failureCount);
//遍历所有的任务
for (ChannelFuture f: futures.values()) {
//判断当前的任务是否失败
if (!f.isSuccess()) {
//如果失败则给集合添加错误的管道对于的错误信息
failed.add(new DefaultEntry<Channel, Throwable>(f.channel(), f.cause()));
}
}
//给当前的任务设置为失败并且传入的异常是个自定义异常，此异常用于存储具体的错误数据信息作为返回值
setFailure0(new ChannelGroupException(failed));
} else {
//如果小于0则代表没有失败那么设置当前的管理是成功状态
setSuccess0();
}
}
}
};
//构造器，所属的ChannelGroup、futures代表当前任务组需要处理的任务们，executor任务执行器
DefaultChannelGroupFuture(ChannelGroup group, Collection<ChannelFuture> futures, EventExecutor executor) {
super(executor);
if (group == null) {
throw new NullPointerException("group");
}
if (futures == null) {
throw new NullPointerException("futures");
}
this.group = group;
//创建一个future集合 key是管道 value是对于的管道任务
Map<Channel, ChannelFuture> futureMap = new LinkedHashMap<Channel, ChannelFuture>();
//将传入的list任务 动态添加到futureMap中
for (ChannelFuture f: futures) {
futureMap.put(f.channel(), f);
}
//因为此任务集合是不允许修改的所以此处转换为了不允许修改的map，此map如果调用put remove等修改方法则会抛出异常
this.futures = Collections.unmodifiableMap(futureMap);
//给管理的所有任务添加任务完成的监听器，传入的任务只要完成就会进入上方的完成监听器从而达到计数的效果
for (ChannelFuture f: this.futures.values()) {
f.addListener(childListener);
}
//如果传入的任务是空则直接完成当前任务，这里可能有些绕 因为是Group他是对多个任务的管理，但是当前类也是一个任务只不过是用于管理其他任务集合的任务罢了
// Done on arrival?
if (this.futures.isEmpty()) {
setSuccess0();
}
}
//此构造器参考上方构造器并没有特殊之处只是减少了将list转map的操作
DefaultChannelGroupFuture(ChannelGroup group, Map<Channel, ChannelFuture> futures, EventExecutor executor) {
super(executor);
this.group = group;
this.futures = Collections.unmodifiableMap(futures);
for (ChannelFuture f: this.futures.values()) {
f.addListener(childListener);
}
// Done on arrival?
if (this.futures.isEmpty()) {
setSuccess0();
}
}
@Override
public ChannelGroup group() {
return group;
}
//根据管道查找任务，就是使用的上方futures的get方法
@Override
public ChannelFuture find(Channel channel) {
return futures.get(channel);
}
//迭代器则是获取的futures的迭代器
@Override
public Iterator<ChannelFuture> iterator() {
return futures.values().iterator();
}
//是否部分成功
@Override
public synchronized boolean isPartialSuccess() {
//successCount != 0 代表总有一个是成功的
//successCount != futures.size() 不是所有的都成功的
return successCount != 0 && successCount != futures.size();
}
//是否部分失败
//参考isPartialSuccess内部解释
@Override
public synchronized boolean isPartialFailure() {
return failureCount != 0 && failureCount != futures.size();
}
//下面的方法都是使用父级的方法此处不做讲解，有疑问可以看父级实现
@Override
public DefaultChannelGroupFuture addListener(GenericFutureListener<? extends Future<? super Void>> listener) {
super.addListener(listener);
return this;
}
@Override
public DefaultChannelGroupFuture addListeners(GenericFutureListener<? extends Future<? super Void>>... listeners) {
super.addListeners(listeners);
return this;
}
@Override
public DefaultChannelGroupFuture removeListener(GenericFutureListener<? extends Future<? super Void>> listener) {
super.removeListener(listener);
return this;
}
@Override
public DefaultChannelGroupFuture removeListeners(
GenericFutureListener<? extends Future<? super Void>>... listeners) {
super.removeListeners(listeners);
return this;
}
@Override
public DefaultChannelGroupFuture await() throws InterruptedException {
super.await();
return this;
}
@Override
public DefaultChannelGroupFuture awaitUninterruptibly() {
super.awaitUninterruptibly();
return this;
}
@Override
public DefaultChannelGroupFuture syncUninterruptibly() {
super.syncUninterruptibly();
return this;
}
@Override
public DefaultChannelGroupFuture sync() throws InterruptedException {
super.sync();
return this;
}
@Override
public ChannelGroupException cause() {
return (ChannelGroupException) super.cause();
}
//需要注意这两个方法是私有方法代表着外部允许设置成功和失败，从而也就看出外部不允许操作这种group组任务的操作
private void setSuccess0() {
super.setSuccess(null);
}
private void setFailure0(ChannelGroupException cause) {
super.setFailure(cause);
}
//上面的讲解这里又得到了验证，如果外部调用了组的设置方法那么将会抛出异常
@Override
public DefaultChannelGroupFuture setSuccess(Void result) {
throw new IllegalStateException();
}
@Override
public boolean trySuccess(Void result) {
throw new IllegalStateException();
}
@Override
public DefaultChannelGroupFuture setFailure(Throwable cause) {
throw new IllegalStateException();
}
@Override
public boolean tryFailure(Throwable cause) {
throw new IllegalStateException();
}
//检查死锁
@Override
protected void checkDeadLock() {
//获取到当前的执行
EventExecutor e = executor();
//如果当前执行器是null自然不会死锁，因为并没有线程执行
//ImmediateEventExecutor是立即执行的执行器还是当前线程所以不会发生死锁，后续再继续线程池介绍的时候将会详细讲解
//e.inEventLoop() 如果当前线程就是执行线程那么就会发生死锁
//这里需要清楚第一第二条件都是派出单线程执行情况，如果是单线程那么无死锁存在，而最后一个条件是判断两个线程，到后面讲解线程池实现时将会讲述
if (e != null && e != ImmediateEventExecutor.INSTANCE && e.inEventLoop()) {
throw new BlockingOperationException();
}
}
//此类很简单用于存储管道和任务
private static final class DefaultEntry<K, V> implements Map.Entry<K, V> {
private final K key;
private final V value;
DefaultEntry(K key, V value) {
this.key = key;
this.value = value;
}
@Override
public K getKey() {
return key;
}
@Override
public V getValue() {
return value;
}
@Override
public V setValue(V value) {
throw new UnsupportedOperationException("read-only");
}
}
}
到此ChannelGroupFuture的实现基本完成了可能有人会觉得一开始的接口并没有用到，这里需要知道一个定义，就是ChannelGroup与ChannelGroupFuture的关系，
ChannelGroup是用于管理管道的而内部很多批量操作最终都是分组成为了ChannelGroupFuture作为统一返回。
