下面列出了怎么用java.nio.channels.NetworkChannel的API类实例代码及写法，或者点击链接到github查看源代码。
public MySQLConnection(NetworkChannel channel, boolean fromSlaveDB, boolean autocommitSynced, boolean isolationSynced) {
super(channel);
this.lastTime = TimeUtil.currentTimeMillis();
this.autocommitSynced = autocommitSynced;
boolean sysAutocommit = DbleServer.getInstance().getConfig().getSystem().getAutocommit() == 1;
this.autocommit = sysAutocommit == autocommitSynced; // T + T-> T, T + F-> F, F +T ->F, F + F->T
this.fromSlaveDB = fromSlaveDB;
this.isolationSynced = isolationSynced;
if (isolationSynced) {
this.txIsolation = DbleServer.getInstance().getConfig().getSystem().getTxIsolation();
} else {
/* if the txIsolation in server.xml is different from the isolation level in MySQL node,
* it need to sync the status firstly for new idle connection*/
this.txIsolation = -1;
}
this.complexQuery = false;
this.usrVariables = new LinkedHashMap<>();
this.sysVariables = new LinkedHashMap<>();
}
@SuppressWarnings({"unchecked", "rawtypes"})
public MySQLConnection make(MySQLDataSource pool, ResponseHandler handler,
String schema) throws IOException {
DataSourceConfig dsc = pool.getConfig();
NetworkChannel channel = openSocketChannel(DbleServer.getInstance().isAIO());
MySQLConnection c = new MySQLConnection(channel, pool.isReadNode(), pool.isAutocommitSynced(), pool.isIsolationSynced());
c.setSocketParams(false);
c.setHost(dsc.getIp());
c.setPort(dsc.getPort());
c.setUser(dsc.getUser());
c.setPassword(dsc.getPassword());
c.setSchema(schema);
c.setHandler(new MySQLConnectionAuthenticator(c, handler));
c.setPool(pool);
c.setIdleTimeout(pool.getConfig().getIdleTimeout());
if (channel instanceof AsynchronousSocketChannel) {
((AsynchronousSocketChannel) channel).connect(
new InetSocketAddress(dsc.getIp(), dsc.getPort()), c,
(CompletionHandler) DbleServer.getInstance().getConnector());
} else {
((NIOConnector) DbleServer.getInstance().getConnector()).postConnect(c);
}
return c;
}
public void setSocketParams(Connection con, boolean isFrontChannel) throws IOException {
int sorcvbuf = 0;
int sosndbuf = 0;
int soNoDelay = 0;
if (isFrontChannel) {
sorcvbuf = netConfig.getFrontsocketsorcvbuf();
sosndbuf = netConfig.getFrontsocketsosndbuf();
soNoDelay = netConfig.getFrontSocketNoDelay();
} else {
sorcvbuf = netConfig.getBacksocketsorcvbuf();
sosndbuf = netConfig.getBacksocketsosndbuf();
soNoDelay = netConfig.getBackSocketNoDelay();
}
NetworkChannel channel = con.getChannel();
channel.setOption(StandardSocketOptions.SO_RCVBUF, sorcvbuf);
channel.setOption(StandardSocketOptions.SO_SNDBUF, sosndbuf);
channel.setOption(StandardSocketOptions.TCP_NODELAY, soNoDelay == 1);
channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
con.setMaxPacketSize(netConfig.getMaxPacketSize());
con.setPacketHeaderSize(netConfig.getPacketHeaderSize());
}
@Override
protected final InetSocketAddress getLocalAddress() throws IOException {
NetworkChannel serverSock = getServerSocket();
if (serverSock == null) {
return null;
}
SocketAddress sa = serverSock.getLocalAddress();
if (sa instanceof InetSocketAddress) {
return (InetSocketAddress) sa;
}
return null;
}
@Override
protected FrontendConnection getConnection(NetworkChannel channel) throws IOException {
ManagerConnection c = new ManagerConnection(channel);
MycatServer.getInstance().getConfig().setSocketParams(c, true);
c.setPrivileges(MycatPrivileges.instance());
c.setQueryHandler(new ManagerQueryHandler(c));
return c;
}
public ServerConnection(NetworkChannel channel)
throws IOException {
super(channel);
this.txInterrupted = false;
this.autocommit = true;
this.preAcStates = true;
this.txReadonly = false;
}
@Override
protected FrontendConnection getConnection(NetworkChannel channel) throws IOException {
SystemConfig sys = MycatServer.getInstance().getConfig().getSystem();
ServerConnection c = new ServerConnection(channel);
MycatServer.getInstance().getConfig().setSocketParams(c, true);
c.setPrivileges(MycatPrivileges.instance());
c.setQueryHandler(new ServerQueryHandler(c));
c.setLoadDataInfileHandler(new ServerLoadDataInfileHandler(c));
c.setPrepareHandler(new ServerPrepareHandler(c,sys.getMaxPreparedStmtCount()));
c.setTxIsolation(sys.getTxIsolation());
c.setSession2(new NonBlockingSession(c));
return c;
}
public void setSocketParams(AbstractConnection con, boolean isFrontChannel)
throws IOException {
int sorcvbuf = 0;
int sosndbuf = 0;
int soNoDelay = 0;
if ( isFrontChannel ) {
sorcvbuf = system.getFrontsocketsorcvbuf();
sosndbuf = system.getFrontsocketsosndbuf();
soNoDelay = system.getFrontSocketNoDelay();
} else {
sorcvbuf = system.getBacksocketsorcvbuf();
sosndbuf = system.getBacksocketsosndbuf();
soNoDelay = system.getBackSocketNoDelay();
}
NetworkChannel channel = con.getChannel();
channel.setOption(StandardSocketOptions.SO_RCVBUF, sorcvbuf);
channel.setOption(StandardSocketOptions.SO_SNDBUF, sosndbuf);
channel.setOption(StandardSocketOptions.TCP_NODELAY, soNoDelay == 1);
channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
con.setMaxPacketSize(system.getMaxPacketSize());
con.setPacketHeaderSize(system.getPacketHeaderSize());
con.setIdleTimeout(system.getIdleTimeout());
con.setCharset(system.getCharset());
}
protected NetworkChannel openSocketChannel(boolean isAIO)
throws IOException {
if (isAIO) {
return AsynchronousSocketChannel
.open(MycatServer.getInstance().getNextAsyncChannelGroup());
} else {
SocketChannel channel = null;
channel = SocketChannel.open();
channel.configureBlocking(false);
return channel;
}
}
public FrontendConnection make(NetworkChannel channel) throws IOException {
channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
FrontendConnection c = getConnection(channel);
MycatServer.getInstance().getConfig().setSocketParams(c, true);
return c;
}
public AbstractConnection(NetworkChannel channel) {
this.channel = channel;
boolean isAIO = (channel instanceof AsynchronousChannel);
if (isAIO) {
socketWR = new AIOSocketWR(this);
} else {
socketWR = new NIOSocketWR(this);
}
this.isClosed = new AtomicBoolean(false);
this.startupTime = TimeUtil.currentTimeMillis();
this.lastReadTime = startupTime;
this.lastWriteTime = startupTime;
}
private void accept(NetworkChannel channel, Long id) {
try {
FrontendConnection c = factory.make(channel);
c.setAccepted(true);
c.setId(id);
NIOProcessor processor = MycatServer.getInstance().nextProcessor();
c.setProcessor(processor);
c.register();
} catch (Exception e) {
LOGGER.error("AioAcceptorError", e);
closeChannel(channel);
}
}
private static void closeChannel(NetworkChannel channel) {
if (channel == null) {
return;
}
try {
channel.close();
} catch (IOException e) {
LOGGER.error("AioAcceptorError", e);
}
}
@SuppressWarnings({ "unchecked", "rawtypes" })
public PostgreSQLBackendConnection make(PostgreSQLDataSource pool,
ResponseHandler handler, final String schema) throws IOException {
final DBHostConfig dsc = pool.getConfig();
NetworkChannel channel = this.openSocketChannel(MycatServer
.getInstance().isAIO());
final PostgreSQLBackendConnection c = new PostgreSQLBackendConnection(
channel, pool.isReadNode());
MycatServer.getInstance().getConfig().setSocketParams(c, false);
// 设置NIOHandler
c.setHandler(new PostgreSQLBackendConnectionHandler(c));
c.setHost(dsc.getIp());
c.setPort(dsc.getPort());
c.setUser(dsc.getUser());
c.setPassword(dsc.getPassword());
c.setSchema(schema);
c.setPool(pool);
c.setResponseHandler(handler);
c.setIdleTimeout(pool.getConfig().getIdleTimeout());
if (channel instanceof AsynchronousSocketChannel) {
((AsynchronousSocketChannel) channel).connect(
new InetSocketAddress(dsc.getIp(), dsc.getPort()), c,
(CompletionHandler) MycatServer.getInstance()
.getConnector());
} else {
((NIOConnector) MycatServer.getInstance().getConnector())
.postConnect(c);
}
return c;
}
public MySQLConnection(NetworkChannel channel, boolean fromSlaveDB) {
super(channel);
this.clientFlags = CLIENT_FLAGS;
this.lastTime = TimeUtil.currentTimeMillis();
this.isQuit = new AtomicBoolean(false);
this.autocommit = true;
this.fromSlaveDB = fromSlaveDB;
// 设为默认值，免得每个初始化好的连接都要去同步一下
this.txIsolation = MycatServer.getInstance().getConfig().getSystem().getTxIsolation();
this.txReadonly = false;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
public MySQLConnection make(MySQLDataSource pool, ResponseHandler handler,
String schema) throws IOException {
DBHostConfig dsc = pool.getConfig();
NetworkChannel channel = openSocketChannel(MycatServer.getInstance()
.isAIO());
MySQLConnection c = new MySQLConnection(channel, pool.isReadNode());
MycatServer.getInstance().getConfig().setSocketParams(c, false);
c.setHost(dsc.getIp());
c.setPort(dsc.getPort());
c.setUser(dsc.getUser());
c.setPassword(dsc.getPassword());
c.setSchema(schema);
c.setHandler(new MySQLConnectionAuthenticator(c, handler));
c.setPool(pool);
c.setIdleTimeout(pool.getConfig().getIdleTimeout());
if (channel instanceof AsynchronousSocketChannel) {
((AsynchronousSocketChannel) channel).connect(
new InetSocketAddress(dsc.getIp(), dsc.getPort()), c,
(CompletionHandler) MycatServer.getInstance()
.getConnector());
} else {
((NIOConnector) MycatServer.getInstance().getConnector())
.postConnect(c);
}
return c;
}
@Override
protected FrontendConnection getConnection(NetworkChannel channel) throws IOException {
ManagerConnection c = new ManagerConnection(channel);
c.setSocketParams(true);
c.setPrivileges(ManagerPrivileges.instance());
c.setHandler(new ManagerAuthenticator(c));
c.setQueryHandler(new ManagerQueryHandler(c));
return c;
}
public ServerConnection(NetworkChannel channel)
throws IOException {
super(channel);
this.txInterrupted = false;
this.autocommit = DbleServer.getInstance().getConfig().getSystem().getAutocommit() == 1;
this.txID = new AtomicLong(1);
this.sptprepare = new ServerSptPrepare(this);
this.usrVariables = new LinkedHashMap<>();
this.sysVariables = new LinkedHashMap<>();
}
@Override
protected FrontendConnection getConnection(NetworkChannel channel) throws IOException {
ServerConnection c = new ServerConnection(channel);
c.setSocketParams(true);
c.setPrivileges(ServerPrivileges.instance());
c.setQueryHandler(new ServerQueryHandler(c));
c.setLoadDataInfileHandler(new ServerLoadDataInfileHandler(c));
c.setPrepareHandler(new ServerPrepareHandler(c));
SystemConfig sys = DbleServer.getInstance().getConfig().getSystem();
c.setTxIsolation(sys.getTxIsolation());
c.setSession2(new NonBlockingSession(c));
return c;
}
protected NetworkChannel openSocketChannel(boolean isAIO)
throws IOException {
if (isAIO) {
return AsynchronousSocketChannel.open(DbleServer.getInstance().getNextAsyncChannelGroup());
} else {
SocketChannel channel = null;
channel = SocketChannel.open();
channel.configureBlocking(false);
return channel;
}
}
public FrontendConnection make(NetworkChannel channel) throws IOException {
channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
FrontendConnection c = getConnection(channel);
c.setSocketParams(true);
return c;
}
public AbstractConnection(NetworkChannel channel) {
this.channel = channel;
boolean isAIO = (channel instanceof AsynchronousChannel);
if (isAIO) {
socketWR = new AIOSocketWR(this);
} else {
socketWR = new NIOSocketWR(this);
}
this.startupTime = TimeUtil.currentTimeMillis();
this.lastReadTime = startupTime;
this.lastWriteTime = startupTime;
}
@Override
protected NetworkChannel getServerSocket() {
return serverSock;
}
@Override
protected NetworkChannel getServerSocket() {
return serverSock;
}
@Override
protected NetworkChannel asNetworkChannel() {
return listening ? serverChannel : clientChannel;
}
@Override
protected NetworkChannel asNetworkChannel() {
return channel;
}
public ManagerConnection(NetworkChannel channel) throws IOException {
super(channel);
}
protected abstract FrontendConnection getConnection(NetworkChannel channel)
throws IOException;
public NetworkChannel getChannel() {
return channel;
}
public BackendAIOConnection(NetworkChannel channel) {
super(channel);
}