下面列出了怎么用io.netty.channel.socket.SocketChannel的API类实例代码及写法，或者点击链接到github查看源代码。
protected void startServer() throws InterruptedException {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new LoggingHandler(LogLevel.DEBUG));
p.addLast(new NettySimpleMessageHandler());
p.addLast(new NettyMasterHandler(DefaultRedisKeeperServer.this, new CommandHandlerManager(), keeperConfig.getTrafficReportIntervalMillis()));
}
});
serverSocketChannel = (ServerSocketChannel) b.bind(currentKeeperMeta.getPort()).sync().channel();
}
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
p.addLast(new EchoServerV4FirstHandler());
p.addLast(new EchoServerV4SecondHandler());
}
});
ChannelFuture f = b.bind(8888).sync();
f.channel().closeFuture().sync();
}
finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
// Enable HTTPS if necessary.
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc()));
}
p.addLast(new HttpClientCodec());
// Remove the following line if you don't want automatic content decompression.
p.addLast(new HttpContentDecompressor());
// Uncomment the following line if you don't want to handle HttpContents.
//p.addLast(new HttpObjectAggregator(1048576));
p.addLast(new HttpSnoopClientHandler());
}
public void initialize() {
logger.info("Initializing the connector");
EventLoopGroup bossGroup = new NioEventLoopGroup(configuration.getIoWorkerThreadCount());
EventLoopGroup workerGroup = new NioEventLoopGroup(configuration.getIoWorkerThreadCount());
bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.childOption(ChannelOption.TCP_NODELAY, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new HttpRequestDecoder(), new HttpResponseEncoder(), new HttpObjectAggregator(configuration.getMaxContentLength()), AbstractHttpConnector.this);
addChannelHandlers(ch.pipeline());
}
});
}
@SneakyThrows
public void start(int port) {
EventLoopGroup group = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(group)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) {
socketChannel.pipeline().addLast(new WriteTimeoutHandler(30), new NettyServerHandler());
}
});
ChannelFuture future = bootstrap.bind().sync();
future.channel().closeFuture().sync();
} finally {
group.shutdownGracefully().sync();
}
}
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
if (sslCtx != null) {
pipeline.addLast("ssl", sslCtx.newHandler(ch.alloc()));
}
pipeline.addLast("codec", new HttpClientCodec());
// Remove the following line if you don't want automatic content decompression.
pipeline.addLast("inflater", new HttpContentDecompressor());
// to be used since huge file transfer
pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
pipeline.addLast("handler", new HttpUploadClientHandler());
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline ph = ch.pipeline();
//入参说明: 读超时时间、写超时时间、所有类型的超时时间、时间格式
ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
// 解码和编码，应和客户端一致
//传输的协议 Protobuf
ph.addLast(new ProtobufVarint32FrameDecoder());
ph.addLast(new ProtobufDecoder(UserMsg.User.getDefaultInstance()));
ph.addLast(new ProtobufVarint32LengthFieldPrepender());
ph.addLast(new ProtobufEncoder());
//业务逻辑实现类
ph.addLast("nettyServerHandler", new NettyServerHandler());
}
private ChannelInitializer getChannelInitializer() throws NoSuchAlgorithmException {
Mac macDummy = Mac.getInstance(controller.getStaticConf().getHmacAlgorithm());
final NettyClientPipelineFactory nettyClientPipelineFactory = new NettyClientPipelineFactory(this, sessionTable,
macDummy.getMacLength(), controller, rl, signatureLength);
ChannelInitializer channelInitializer = new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(nettyClientPipelineFactory.getDecoder());
ch.pipeline().addLast(nettyClientPipelineFactory.getEncoder());
ch.pipeline().addLast(nettyClientPipelineFactory.getHandler());
}
};
return channelInitializer;
}
public static void main(String[] args) throws Exception {
String ip = "127.0.0.1";
int port = 8080;
// Configure SSL.
SelfSignedCertificate ssc = new SelfSignedCertificate();
final SslContext sslCtx = SslContext.newServerContext(
ssc.certificate(), ssc.privateKey(), null, null,
IdentityCipherSuiteFilter.INSTANCE,
new ApplicationProtocolConfig(Protocol.ALPN,
SelectorFailureBehavior.FATAL_ALERT,
SelectedListenerFailureBehavior.FATAL_ALERT,
SelectedProtocol.SPDY_3_1.protocolName(),
SelectedProtocol.HTTP_1_1.protocolName()), 0, 0);
ChannelInitializer<SocketChannel> channelInit = new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(sslCtx.newHandler(ch.alloc()));
p.addLast(new SpdyOrHttpHandler());
}
};
NettyServerUtil.newHttpServerBootstrap(ip, port, channelInit);
}
public void run() throws Exception {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap(); // (2)
b.group(workerGroup)
.channel(NioSocketChannel.class) // (3)
.handler(new ChannelInitializer<SocketChannel>() { // (4)
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(clientHandler);
}
})
.option(ChannelOption.SO_KEEPALIVE, true); // (6)
// Bind and start to accept incoming connections.
ChannelFuture f = b.connect("localhost", port).sync(); // (7)
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
@Override
protected void createConnectionInternal(NetNode node, AsyncConnectionManager connectionManager,
AsyncCallback<AsyncConnection> ac) {
final InetSocketAddress inetSocketAddress = node.getInetSocketAddress();
bootstrap.connect(node.getHost(), node.getPort()).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
SocketChannel ch = (SocketChannel) future.channel();
NettyWritableChannel writableChannel = new NettyWritableChannel(ch);
AsyncConnection conn;
if (connectionManager != null) {
conn = connectionManager.createConnection(writableChannel, false);
} else {
conn = new TcpClientConnection(writableChannel, NettyNetClient.this);
}
ch.pipeline().addLast(new NettyNetClientHandler(NettyNetClient.this, connectionManager, conn));
conn.setInetSocketAddress(inetSocketAddress);
conn = NettyNetClient.this.addConnection(inetSocketAddress, conn);
ac.setAsyncResult(conn);
} else {
ac.setAsyncResult(future.cause());
}
}
});
}
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
if (sslCtx != null) {
pipeline.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
}
pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, new ByteBuf[]{
Unpooled.wrappedBuffer(Constant.RPC_DELIMITER.getBytes())
}));
pipeline.addLast(DECODER);
pipeline.addLast(ENCODER);
// and then unit logic.
pipeline.addLast(new RpcClientHandler(nodeId)/*CLIENT_HANDLER*/);
pipeline.addLast(new RpcClientDecoder());
pipeline.addLast(new StreamRpcClientHandler());
pipeline.addLast(new RpcClientUnitHandler());
}
@Override
protected void initChannel(SocketChannel e) throws Exception {
//protobuf处理器
// e.pipeline().addLast("frameDecoder", new ProtobufVarint32FrameDecoder());//用于Decode前解决半包和粘包问题
//此处需要传入自定义的protobuf的defaultInstance
// e.pipeline().addLast("protobufDecoder",new ProtobufDecoder();
//将请求和应答消息解码为http消息
e.pipeline().addLast("http-codec", new HttpServerCodec());
//HttpObjectAggregator:将Http消息的多个部分合成一条完整的消息
e.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
//ChunkedWriteHandler:向客户端发送HTML5文件
e.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
//在管道中添加我们自己实现的接收数据实现方法
e.pipeline().addLast("handler", new WebSocketServerHandler());
}
public void run() throws InterruptedException {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new DelimiterBasedFrameDecoder(MAX_FRAME_LENGTH, false, Delimiters.lineDelimiter()));
p.addLast(new CarbonServerHandler(bus, configuration.getCarbon().getBaseRollup()));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
logger.error(cause);
super.exceptionCaught(ctx, cause);
}
});
// Start the server.
b.bind(configuration.getCarbon().getBind(), configuration.getCarbon().getPort()).sync();
}
@PostConstruct
public void start() throws Exception {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workGroup)
.channelFactory(NioServerSocketChannel::new)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
pipeline.addLast("handler", nodeChannelInBoundHandler);
}
})
.option(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_SNDBUF, 2048)
.option(ChannelOption.SO_RCVBUF, 1024);
bootstrap.bind(goPushNodeServerConfig.getNodePort()).sync();
log.info("Node server start successful! listening port: {}", goPushNodeServerConfig.getNodePort());
}
public NettyIoAcceptor(NettyIoServiceFactory factory, IoHandler handler) {
this.factory = factory;
this.handler = handler;
channelGroup = new DefaultChannelGroup("sshd-acceptor-channels", GlobalEventExecutor.INSTANCE);;
bootstrap.group(factory.eventLoopGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new NettyIoSession(NettyIoAcceptor.this, handler).adapter);
}
});
}
public static void newServerBootstrap(String host, int port, ChannelInitializer<SocketChannel> initializer){
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(initializer)
.bind(port).sync().channel().closeFuture().sync();
} catch (Exception e){
e.printStackTrace();
}
finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
@Override
public ChannelInitializer<SocketChannel> newChannelInitializer() {
return new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(loggingHandler)
.addLast(new IdleStateHandler(30, 0, 10))
.addLast("decoder", new HttpRequestDecoder())
.addLast("http-aggregator", new HttpObjectAggregator(65536))
.addLast("encoder", new HttpResponseEncoder())
.addLast("chunk", new ChunkedWriteHandler())
.addLast(businessExecutors, "business-handler", new HttpServerHandler())
.addLast(new HeartbeatHandler())
.addLast(exceptionHandler);
}
};
}
public void remove(SocketChannel socketChannel){
for (@SuppressWarnings("rawtypes") Map.Entry entry:map.entrySet()){
if (entry.getValue()==socketChannel){
log.info("#############客户端下线##############");
log.info("下线主机名为："+entry.getKey());
Client client=clientService.selectClientByClientIP(entry.getKey().toString());
if(client!=null)
{
client.setStatus(1);
clientMapper.updateClient(client);
//登陆失败，删除心跳map中的数据
NettyServer.clientMap.remove(client.getClientIp());
}
map.remove(entry.getKey());
socketChannel.close();
}
}
}
@Before
public void setup() throws Exception {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new HttpClientCodec());
p.addLast(new HttpContentDecompressor());
p.addLast(new SimpleChannelInboundHandler<HttpObject>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
response = prepareResponse(ctx, msg, response);
}
});
}
});
channel = b.connect(HOST, PORT)
.sync()
.channel();
}
private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, Config conf) {
LOG.debug("Configuring Websocket MQTT transport");
if(!conf.websocketEnabled) {
return;
}
int port = conf.websocketPort;// Integer.parseInt(webSocketPortProp);
final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
String host = conf.websocketHost;
initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
@Override
void init(SocketChannel channel) {
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
pipeline.addLast("webSocketHandler",
new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
configureMQTTPipeline(pipeline, timeoutHandler, handler);
}
});
}
public void start() throws Exception{
NioEventLoopGroup group = new NioEventLoopGroup();
ServerBootstrap strap = new ServerBootstrap();
try {
strap.group(group).channel(NioServerSocketChannel.class).localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel sc) throws Exception{
sc.pipeline().addLast(new EchoServerHandler());
}
});
ChannelFuture future = strap.bind().sync();
System.out.println(EchoServer.class.getName() + " started and listen on " + future.channel().localAddress());
future.channel().closeFuture().sync();
} finally {
group.shutdownGracefully().sync();
}
}
public NettyIoAcceptor(NettyIoServiceFactory factory, final IoHandler handler) {
this.factory = factory;
this.handler = handler;
channelGroup = new DefaultChannelGroup("sshd-acceptor-channels", GlobalEventExecutor.INSTANCE);
bootstrap.group(factory.eventLoopGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new NettyIoSession(NettyIoAcceptor.this, handler).adapter);
}
});
}
@Override
public void create() {
Gdx.app.setLogLevel(Application.LOG_DEBUG);
bossGroup = new NioEventLoopGroup();
workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap()
.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
TcpEndpoint endpoint = new TcpEndpoint(ch, Main.this);
Main.this.endpoint = endpoint;
ch.pipeline()
.addFirst(connectionLimiter)
.addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
notifyChannelInactive(ctx);
}
})
.addLast(new SizePrefixedDecoder())
.addLast(new EndpointedChannelHandler<>(ByteBuf.class, endpoint))
;
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture f = b.bind(PORT).sync();
} catch (Throwable t) {
Gdx.app.error(TAG, t.getMessage(), t);
Gdx.app.exit();
}
}
@Override
protected void initChannel(final SocketChannel ch) {
//及时发送 与 缓存发送
Channel channel = new NettyChannel(ch, true);
//设置payload,添加业务线程池到channel
channel.setAttribute(Channel.PAYLOAD, url.getPositiveInt(Constants.PAYLOAD))
.setAttribute(Channel.BIZ_THREAD_POOL, bizThreadPool, (k, v) -> v != null);
if (sslContext != null) {
ch.pipeline().addFirst("ssl", sslContext.newHandler(ch.alloc()));
}
ch.pipeline().addLast("connection", new ConnectionChannelHandler(channel, publisher) {
@Override
public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
removeChannel(channel);
super.channelInactive(ctx);
logger.info(String.format("disconnect %s", ctx.channel().remoteAddress()));
}
});
if (adapter != null) {
ch.pipeline().addLast("adapter", new ProtocolAdapterDecoder(adapter, channel));
} else {
AdapterContext context = new ProtocolAdapterContext(channel, ch.pipeline());
context.bind(codec, chain);
}
ChannelTransport transport = function.apply(channel, url);
channel.setAttribute(Channel.CHANNEL_TRANSPORT, transport);
channel.setAttribute(Channel.SERVER_CHANNEL, getServerChannel());
addChannel(channel, transport);
}
@Override
protected void initSocketChannel(SocketChannel ch) {
super.initSocketChannel(ch);
ch.pipeline().addLast(new HttpServerCodec());
ch.pipeline().addLast(new HttpObjectAggregator(65536));
ch.pipeline().addLast(new ChunkedWriteHandler());
HttpServerHandler httpServerHandler = new HttpServerHandler(this, httpResource);
processHttpHandler(httpServerHandler);
ch.pipeline().addLast("http", httpServerHandler);
}
@Override
public void initChannel(SocketChannel ch) throws Exception {
settingsHandler = new Http2SettingsHandler(ch.newPromise());
responseHandler = new Http2ClientResponseHandler();
if (sslCtx != null) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(sslCtx.newHandler(ch.alloc(), host, port));
pipeline.addLast(Http2Util.getClientAPNHandler(maxContentLength, settingsHandler, responseHandler));
}
}
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
if (sslCtx != null) {
pipeline.addLast(sslCtx.newHandler(ch.alloc()));
}
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
pipeline.addLast(new WebSocketServerCompressionHandler());
pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, subProtocols, true));
pipeline.addLast(new WebSocketRemoteServerFrameHandler());
}
public static void newHttpServerBootstrap(String ip, int port, ChannelInitializer<SocketChannel> channelInitializer){
// Configure the server.
int numCPUs = Runtime.getRuntime().availableProcessors();
EventLoopGroup bossGroup = new NioEventLoopGroup(numCPUs);
EventLoopGroup workerGroup = new NioEventLoopGroup(numCPUs);
try {
//public service processor
ServerBootstrap publicServerBootstrap = new ServerBootstrap();
publicServerBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);
publicServerBootstrap.childOption(ChannelOption.TCP_NODELAY, false)
.childOption(ChannelOption.SO_KEEPALIVE, false)
.childHandler(channelInitializer);
//bind to public access host info
Channel ch1;
if("*".equals(ip)){
ch1 = publicServerBootstrap.bind(port).sync().channel();
} else {
ch1 = publicServerBootstrap.bind(ip, port).sync().channel();
}
System.out.println(String.format("Started OK HttpServer at %s:%d", ip, port));
ch1.config().setConnectTimeoutMillis(1800);
ch1.closeFuture().sync();
System.out.println("Shutdown...");
} catch (Throwable e) {
e.printStackTrace();
System.exit(1);
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
/**
* 构造
* @param channelInitializer {@link ChannelInitializer}
* @param isDebug
*/
public ANetty(ChannelInitializer<SocketChannel> channelInitializer, boolean isDebug){
this.isDebug = isDebug;
this.mChannelInitializer = channelInitializer;
initHandlerThread();
mHandler.sendEmptyMessage(NETTY_INIT);
}