序
本文主要研究一下reactor-netty中TcpClient的create的过程
maven
<dependency>
<groupId>io.projectreactor.ipc</groupId>
<artifactId>reactor-netty</artifactId>
<version>0.7.3.RELEASE</version>
</dependency>
TcpClient
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/tcp/TcpClient.java
protected TcpClient(TcpClient.Builder builder) {
ClientOptions.Builder<?> clientOptionsBuilder = ClientOptions.builder();
if (Objects.nonNull(builder.options)) {
builder.options.accept(clientOptionsBuilder);
}
if (!clientOptionsBuilder.isLoopAvailable()) {
clientOptionsBuilder.loopResources(TcpResources.get());
}
if (!clientOptionsBuilder.isPoolAvailable() && !clientOptionsBuilder.isPoolDisabled()) {
clientOptionsBuilder.poolResources(TcpResources.get());
}
this.options = clientOptionsBuilder.build();
}
loopResources和poolResources其实是通过TcpResources创建 上面loopResources创建完之后，下面的poolResources其实是直接返回
clientOptionsBuilder.isLoopAvailable()
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/options/NettyOptions.java
public final boolean isLoopAvailable() {
return this.loopResources != null;
}
一开始是null，于是调用TcpResources.get()创建
TcpResources.get()
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/tcp/TcpResources.java
/**
* Return the global HTTP resources for event loops and pooling
*
* @return the global HTTP resources for event loops and pooling
*/
public static TcpResources get() {
return getOrCreate(tcpResources, null, null, ON_TCP_NEW, "tcp");
}
/**
* Safely check if existing resource exist and proceed to update/cleanup if new
* resources references are passed.
*
* @param ref the resources atomic reference
* @param loops the eventual new {@link LoopResources}
* @param pools the eventual new {@link PoolResources}
* @param onNew a {@link TcpResources} factory
* @param name a name for resources
* @param <T> the reified type of {@link TcpResources}
*
* @return an existing or new {@link TcpResources}
*/
protected static <T extends TcpResources> T getOrCreate(AtomicReference<T> ref,
LoopResources loops,
PoolResources pools,
BiFunction<LoopResources, PoolResources, T> onNew,
String name) {
T update;
for (; ; ) {
T resources = ref.get();
if (resources == null || loops != null || pools != null) {
update = create(resources, loops, pools, name, onNew);
if (ref.compareAndSet(resources, update)) {
if(resources != null){
if(loops != null){
resources.defaultLoops.dispose();
}
if(pools != null){
resources.defaultPools.dispose();
}
}
return update;
}
else {
update._dispose();
}
}
else {
return resources;
}
}
}
这里进入create，创建loops还有pools
static final AtomicReference<TcpResources> tcpResources;
static final BiFunction<LoopResources, PoolResources, TcpResources> ON_TCP_NEW;
static {
ON_TCP_NEW = TcpResources::new;
tcpResources = new AtomicReference<>();
}
final PoolResources defaultPools;
final LoopResources defaultLoops;
protected TcpResources(LoopResources defaultLoops, PoolResources defaultPools) {
this.defaultLoops = defaultLoops;
this.defaultPools = defaultPools;
}
static <T extends TcpResources> T create(T previous,
LoopResources loops,
PoolResources pools,
String name,
BiFunction<LoopResources, PoolResources, T> onNew) {
if (previous == null) {
loops = loops == null ? LoopResources.create("reactor-" + name) : loops;
pools = pools == null ? PoolResources.elastic(name) : pools;
}
else {
loops = loops == null ? previous.defaultLoops : loops;
pools = pools == null ? previous.defaultPools : pools;
}
return onNew.apply(loops, pools);
}
这里的onNew是创建TcpResources，使用的构造器是TcpResources(LoopResources defaultLoops, PoolResources defaultPools)
LoopResources.create
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/LoopResources.java
/**
* Default worker thread count, fallback to available processor
*/
int DEFAULT_IO_WORKER_COUNT = Integer.parseInt(System.getProperty(
"reactor.ipc.netty.workerCount",
"" + Math.max(Runtime.getRuntime()
.availableProcessors(), 4)));
/**
* Default selector thread count, fallback to -1 (no selector thread)
*/
int DEFAULT_IO_SELECT_COUNT = Integer.parseInt(System.getProperty(
"reactor.ipc.netty.selectCount",
"" + -1));
/**
* Create a simple {@link LoopResources} to provide automatically for {@link
* EventLoopGroup} and {@link Channel} factories
*
* @param prefix the event loop thread name prefix
*
* @return a new {@link LoopResources} to provide automatically for {@link
* EventLoopGroup} and {@link Channel} factories
*/
static LoopResources create(String prefix) {
return new DefaultLoopResources(prefix, DEFAULT_IO_SELECT_COUNT,
DEFAULT_IO_WORKER_COUNT,
true);
}
这里有两个参数，一个是worker thread count，一个是selector thread count
- DEFAULT_IO_WORKER_COUNT
如果环境变量有设置reactor.ipc.netty.workerCount，则用该值；没有设置则取Math.max(Runtime.getRuntime().availableProcessors(), 4)))
- DEFAULT_IO_SELECT_COUNT
如果环境变量有设置reactor.ipc.netty.selectCount，则用该值；没有设置则取-1，表示没有selector thread
DefaultLoopResources
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/DefaultLoopResources.java
DefaultLoopResources(String prefix,
int selectCount,
int workerCount,
boolean daemon) {
this.running = new AtomicBoolean(true);
this.daemon = daemon;
this.workerCount = workerCount;
this.prefix = prefix;
this.serverLoops = new NioEventLoopGroup(workerCount,
threadFactory(this, "nio"));
this.clientLoops = LoopResources.colocate(serverLoops);
this.cacheNativeClientLoops = new AtomicReference<>();
this.cacheNativeServerLoops = new AtomicReference<>();
if (selectCount == -1) {
this.selectCount = workerCount;
this.serverSelectLoops = this.serverLoops;
this.cacheNativeSelectLoops = this.cacheNativeServerLoops;
}
else {
this.selectCount = selectCount;
this.serverSelectLoops =
new NioEventLoopGroup(selectCount, threadFactory(this, "select-nio"));
this.cacheNativeSelectLoops = new AtomicReference<>();
}
}
这里prefix为reactor-tcp，selectCount为-1，workerCount为4，daemon为true 可以看到这里创建了NioEventLoopGroup，workerCount为4；由于selectCount=-1因此没有单独创建serverSelectLoops
NioEventLoopGroup
netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/nio/NioEventLoopGroup.java
public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory,
final SelectorProvider selectorProvider, final SelectStrategyFactory selectStrategyFactory) {
super(nThreads, threadFactory, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
}
注意这里的rejectHandler是RejectedExecutionHandlers.reject()
netty-common-4.1.20.Final-sources.jar!/io/netty/util/concurrent/MultithreadEventExecutorGroup.java
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param threadFactory the ThreadFactory to use, or {@code null} if the default should be used.
* @param args arguments which will passed to each {@link #newChild(Executor, Object...)} call
*/
protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
this(nThreads, threadFactory == null ? null : new ThreadPerTaskExecutor(threadFactory), args);
}
new NioEventLoopGroup的时候调用了MultithreadEventExecutorGroup 这里的threadFactory是reactor.ipc.netty.resources.DefaultLoopResources$EventLoopSelectorFactory 这里的executor是ThreadPerTaskExecutor
netty-common-4.1.20.Final-sources.jar!/io/netty/util/concurrent/ThreadPerTaskExecutor.java
public final class ThreadPerTaskExecutor implements Executor {
private final ThreadFactory threadFactory;
public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
if (threadFactory == null) {
throw new NullPointerException("threadFactory");
}
this.threadFactory = threadFactory;
}
@Override
public void execute(Runnable command) {
threadFactory.newThread(command).start();
}
}
MultithreadEventExecutorGroup
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param executor the Executor to use, or {@code null} if the default should be used.
* @param chooserFactory the {@link EventExecutorChooserFactory} to use.
* @param args arguments which will passed to each {@link #newChild(Executor, Object...)} call
*/
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
if (nThreads <= 0) {
throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
}
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(executor, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
if (!success) {
for (int j = 0; j < i; j ++) {
children[j].shutdownGracefully();
}
for (int j = 0; j < i; j ++) {
EventExecutor e = children[j];
try {
while (!e.isTerminated()) {
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
}
} catch (InterruptedException interrupted) {
// Let the caller handle the interruption.
Thread.currentThread().interrupt();
break;
}
}
}
}
}
chooser = chooserFactory.newChooser(children);
final FutureListener<Object> terminationListener = new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> future) throws Exception {
if (terminatedChildren.incrementAndGet() == children.length) {
terminationFuture.setSuccess(null);
}
}
};
for (EventExecutor e: children) {
e.terminationFuture().addListener(terminationListener);
}
Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet, children);
readonlyChildren = Collections.unmodifiableSet(childrenSet);
}
注意，这里用for循环去newChild
netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/nio/NioEventLoopGroup.java
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
return new NioEventLoop(this, executor, (SelectorProvider) args[0],
((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
每个child都是一个NioEventLoop
NioEventLoop
netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/nio/NioEventLoop.java
protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16,
SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE));
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
if (selectorProvider == null) {
throw new NullPointerException("selectorProvider");
}
if (strategy == null) {
throw new NullPointerException("selectStrategy");
}
provider = selectorProvider;
final SelectorTuple selectorTuple = openSelector();
selector = selectorTuple.selector;
unwrappedSelector = selectorTuple.unwrappedSelector;
selectStrategy = strategy;
}
注意这里的DEFAULT_MAX_PENDING_TASKS参数，指定了队列的大小。 如果io.netty.eventLoop.maxPendingTasks有指定，则取它跟16的最大值；没有指定则是Integer.MAX_VALUE 这里没有指定，默认是Integer.MAX_VALUE
NioEventLoop extends SingleThreadEventLoop
netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/SingleThreadEventLoop.java
protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
boolean addTaskWakesUp, int maxPendingTasks,
RejectedExecutionHandler rejectedExecutionHandler) {
super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);
tailTasks = newTaskQueue(maxPendingTasks);
}
这里的parent是NioEventLoopGroup 这里的executor是ThreadPerTaskExecutor 这里的rejectHandler是RejectedExecutionHandlers.reject()
SingleThreadEventLoop extends SingleThreadEventExecutor
/**
* Create a new instance
*
* @param parent the {@link EventExecutorGroup} which is the parent of this instance and belongs to it
* @param executor the {@link Executor} which will be used for executing
* @param addTaskWakesUp {@code true} if and only if invocation of {@link #addTask(Runnable)} will wake up the
* executor thread
* @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
* @param rejectedHandler the {@link RejectedExecutionHandler} to use.
*/
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
boolean addTaskWakesUp, int maxPendingTasks,
RejectedExecutionHandler rejectedHandler) {
super(parent);
this.addTaskWakesUp = addTaskWakesUp;
this.maxPendingTasks = Math.max(16, maxPendingTasks);
this.executor = ObjectUtil.checkNotNull(executor, "executor");
taskQueue = newTaskQueue(this.maxPendingTasks);
rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
}
/**
* Create a new {@link Queue} which will holds the tasks to execute. This default implementation will return a
* {@link LinkedBlockingQueue} but if your sub-class of {@link SingleThreadEventExecutor} will not do any blocking
* calls on the this {@link Queue} it may make sense to {@code @Override} this and return some more performant
* implementation that does not support blocking operations at all.
*/
protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
return new LinkedBlockingQueue<Runnable>(maxPendingTasks);
}
这里的maxPendingTasks是Integer.MAX_VALUE，创建的taskQueue的大小为Integer.MAX_VALUE 这里的addTaskWakesUp为false
PoolResources.elastic(name)
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/PoolResources.java
/**
* Create an uncapped {@link PoolResources} to provide automatically for {@link
* ChannelPool}.
* <p>An elastic {@link PoolResources} will never wait before opening a new
* connection. The reuse window is limited but it cannot starve an undetermined volume
* of clients using it.
*
* @param name the channel pool map name
*
* @return a new {@link PoolResources} to provide automatically for {@link
* ChannelPool}
*/
static PoolResources elastic(String name) {
return new DefaultPoolResources(name, SimpleChannelPool::new);
}
DefaultPoolResources
reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/DefaultPoolResources.java
final ConcurrentMap<SocketAddress, Pool> channelPools;
final String name;
final PoolFactory provider;
DefaultPoolResources(String name, PoolFactory provider) {
this.name = name;
this.provider = provider;
this.channelPools = PlatformDependent.newConcurrentHashMap();
}
创建channelPools的map，key是SocketAddress，value是Pool
小结
TcpClient的create方法主要是创建TcpResources，而TcpResources则创建loopResources和poolResources。
loopResources
这个loopResources主要是创建NioEventLoopGroup，以及该group下面的workerCount个NioEventLoop(这里涉及两个参数，一个是worker thread count，一个是selector thread count)
- DEFAULT_IO_WORKER_COUNT：如果环境变量有设置reactor.ipc.netty.workerCount，则用该值；没有设置则取Math.max(Runtime.getRuntime().availableProcessors(), 4)))
- DEFAULT_IO_SELECT_COUNT：如果环境变量有设置reactor.ipc.netty.selectCount，则用该值；没有设置则取-1，表示没有selector thread
- DEFAULT_MAX_PENDING_TASKS: 指定NioEventLoop的taskQueue的大小，Math.max(16,SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE))
- NioEventLoop继承了SingleThreadEventLoop，而SingleThreadEventLoop则继承SingleThreadEventExecutor，而其代理的executor是ThreadPerTaskExecutor，rejectHandler是RejectedExecutionHandlers.reject()，默认的taskQueue是LinkedBlockingQueue，其大小为Integer.MAX_VALUE
poolResources
这个主要是创建channelPools，类型是ConcurrentMap<SocketAddress, Pool>
