客户端建立连接环节：（线程就像车的引擎，是RPC框架设计的关键）
1.主线程（main）初始化2个对象：ChannelFactory 和ChannelPipelineFactory
（1）初始化NioClientSocketChannelFactory的3个属性：bossPool ，workerPool ，sink （new NioClientSocketPipelineSink(bossPool)）并启动boss和工作线程（见文章后面的第5点）
（2）初始化DefaultChannelPipeline的属性：name2ctx，tail，
2.主线程（main）通过 clientbootstrap.connect(socketAddress)初始化NioClientSocketChannel的work线程，pipeline，sink：
由此可见：NETTY3.X是一个NioClientSocketChannel一个work线程。
public ChannelFuture connect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
if (remoteAddress == null) {
throw new NullPointerException("remoteAddress");
}
ChannelPipeline pipeline;
try {
pipeline = getPipelineFactory().getPipeline();
} catch (Exception e) {
throw new ChannelPipelineException("Failed to initialize a pipeline.", e);
}
// Set the options.
Channel ch = getFactory().newChannel(pipeline);
boolean success = false;
try {
ch.getConfig().setOptions(getOptions());
success = true;
} finally {
if (!success) {
ch.close();
}
}
// Bind.
if (localAddress != null) {
ch.bind(localAddress);
}
// Connect.
return ch.connect(remoteAddress);
}
public SocketChannel newChannel(ChannelPipeline pipeline) {
return new NioClientSocketChannel(this, pipeline, sink, workerPool.nextWorker());
}
3.最终经过设定好的DefaultChannelPipeline的handler处理后交给NioClientSocketPipelineSink处理事件：OPEN,CONNECTED：
public void eventSunk(
ChannelPipeline pipeline, ChannelEvent e) throws Exception {
if (e instanceof ChannelStateEvent) {
ChannelStateEvent event = (ChannelStateEvent) e;
NioClientSocketChannel channel =
(NioClientSocketChannel) event.getChannel();
ChannelFuture future = event.getFuture();
ChannelState state = event.getState();
Object value = event.getValue();
switch (state) {
case OPEN:
if (Boolean.FALSE.equals(value)) {
channel.worker.close(channel, future);
}
break;
case BOUND:
if (value != null) {
bind(channel, future, (SocketAddress) value);
} else {
channel.worker.close(channel, future);
}
break;
case CONNECTED:
if (value != null) {
connect(channel, future, (SocketAddress) value);
} else {
channel.worker.close(channel, future);
}
break;
case INTEREST_OPS:
channel.worker.setInterestOps(channel, future, ((Integer) value).intValue());
break;
}
} else if (e instanceof MessageEvent) {
MessageEvent event = (MessageEvent) e;
NioSocketChannel channel = (NioSocketChannel) event.getChannel();
boolean offered = channel.writeBufferQueue.offer(event);
assert offered;
channel.worker.writeFromUserCode(channel);
}
}
4.注册事件：
（1）注册NioClientSocketChannel的OP_READ事件到nioworker线程的selector上进行监听
nioworker类：
private final class RegisterTask implements Runnable {
private final NioSocketChannel channel;
private final ChannelFuture future;
private final boolean server;
RegisterTask(
NioSocketChannel channel, ChannelFuture future, boolean server) {
this.channel = channel;
this.future = future;
this.server = server;
}
public void run() {
SocketAddress localAddress = channel.getLocalAddress();
SocketAddress remoteAddress = channel.getRemoteAddress();
if (localAddress == null || remoteAddress == null) {
if (future != null) {
future.setFailure(new ClosedChannelException());
}
close(channel, succeededFuture(channel));
return;
}
try {
if (server) {
channel.channel.configureBlocking(false);
}
channel.channel.register(
selector, channel.getInternalInterestOps(), channel);
if (future != null) {
channel.setConnected();
future.setSuccess();
}
if (server || !((NioClientSocketChannel) channel).boundManually) {
fireChannelBound(channel, localAddress);
}
fireChannelConnected(channel, remoteAddress);
} catch (IOException e) {
if (future != null) {
future.setFailure(e);
}
close(channel, succeededFuture(channel));
if (!(e instanceof ClosedChannelException)) {
throw new ChannelException(
"Failed to register a socket to the selector.", e);
}
}
}
}
（2）注册NioClientSocketChannel的OP_CONNECT事件到NIOclientBoss线程的selector上进行监听
NioClientBoss类：
private final class RegisterTask implements Runnable {
private final NioClientBoss boss;
private final NioClientSocketChannel channel;
RegisterTask(NioClientBoss boss, NioClientSocketChannel channel) {
this.boss = boss;
this.channel = channel;
}
public void run() {
int timeout = channel.getConfig().getConnectTimeoutMillis();
if (timeout > 0) {
if (!channel.isConnected()) {
channel.timoutTimer = timer.newTimeout(wakeupTask,
timeout, TimeUnit.MILLISECONDS);
}
}
try {
channel.channel.register(
boss.selector, SelectionKey.OP_CONNECT, channel);
} catch (ClosedChannelException e) {
channel.worker.close(channel, succeededFuture(channel));
}
int connectTimeout = channel.getConfig().getConnectTimeoutMillis();
if (connectTimeout > 0) {
channel.connectDeadlineNanos = System.nanoTime() + connectTimeout * 1000000L;
}
}
}
5.NioClientSocketChannelFactory启动的IO线程：(1个boss多个work线程)
初始化线程池：
private static ExecutorService getCachedExecutor(String name) {
//newCachedThreadPool:根据需要创建线程，需考虑内存是否够用，理论上一个线程1M内存，存在内存溢出的风险
return Executors.newCachedThreadPool(new NamedThreadFactory(name));
}
public AbstractHsfService() {
this(getCachedExecutor("HSF-BOSS-PROCESSOR"), Runtime.getRuntime().availableProcessors() + 1);
}
public AbstractHsfService(Executor bossExecutor, int workerCount) {
this(bossExecutor, getCachedExecutor("HSF-WORKER-PROCESSOR"), workerCount);
}
public AbstractHsfService(Executor bossExecutor, Executor workerExecutor, int workerCount) {
if (bossExecutor == null) {
throw new IllegalArgumentException("bossExecutor can not be null.");
} else if (workerExecutor == null) {
throw new IllegalArgumentException("workerExecutor can not be null.");
} else if (workerCount <= 0) {
throw new IllegalArgumentException("workerCount required > 0.");
} else if (workerExecutor instanceof ThreadPoolExecutor
&& ((ThreadPoolExecutor) workerExecutor).getMaximumPoolSize() < workerCount) {
throw new IllegalArgumentException("the maximum pool size of workerExecutor required >= workerCount.");
}
this.bossExecutor = bossExecutor;
this.workerExecutor = workerExecutor;
this.workerCount = workerCount;
init();
}
启动工作线程：AbstractNioWorkerPool类
protected void init() {
if (!initialized.compareAndSet(false, true)) {
throw new IllegalStateException("initialized already");
}
for (int i = 0; i < workers.length; i++) {
workers[i] = newWorker(workerExecutor);
}
waitForWorkerThreads();
}
启动BOSS线程：AbstractNioBossPool类
protected void init() {
if (!initialized.compareAndSet(false, true)) {
throw new IllegalStateException("initialized already");
}
for (int i = 0; i < bosses.length; i++) {
bosses[i] = newBoss(bossExecutor);
}
waitForBossThreads();
}
6.在DispatchUpStreamHandler中初始化封装的HsfChannel：
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
logger.warn("channel {} is connected.", e.getChannel());
//
HsfChannel hsfChannel = new HsfChannel(eventDispatcher.getService(), e.getChannel());
ChannelGroupFuture channelFuture = ChannelGroupFutureHolder.remove(hsfChannel.getId());
hsfChannel.setChannelGroupFuture(channelFuture);
//
eventDispatcher.getService().getChannels().put(hsfChannel.getId(), hsfChannel);
eventDispatcher.dispatchChannelEvent(ctx, hsfChannel, e);
super.channelConnected(ctx, e);
}
- 大小: 17.7 KB
分享到：
相关推荐
笔者工作的这几年之中，总结并开发了如下几个框架： summercool（Web 框架，已经应用于某国内大型网络公司的等重要应用）、summercool-hsf（基于Netty实现的RPC框架，已经应用国内某移动互联网公司）、 summercool-...
summercool-hsf Automatically exported from code.google.com/p/summercool-hsf 1.目前为止性能最高的RPC远程通讯框架 2.也可以做为手机长连接的Server，经测试已经达到了50W以上的性能长连接 （需调整linux内核...
依赖Xml代码 收藏代码org.summercoolsummercool-ddl1.0源码svn地址：2.准备Sql映射文件Xml代码 收藏代码 insert into $[user]$( id, name ) values ( #{id}, #{name} ) 注意此处，对于需要分表或分库的表，使用“$...
summercool-ddl Automatically exported from code.google.com/p/summercool-ddl 学习了解使用！
笔者工作的这几年之中，总结并开发了如下几个框架： summercool（ Web框架，已经应用于某国内大型网络公司的等重要应用）、summercool-hsf（基于Netty实现的RPC框架，已经应用国内某移动互联网公司）、summercool-...
在IT行业中，数据库管理和优化是至关重要的环节，尤其是在大数据量的场景下。本文将深入探讨"Ibatis之分表分库解决方案"，这是一个针对高并发、大数据处理的有效策略。Ibatis，作为一个轻量级的Java持久层框架，允许...