Question

请教一个关于netty的问题

Answer 1

如果是proxy，那么netty就有一个例子了

import java.net.InetSocketAddress;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

/**
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
*/
public class HexDumpProxy {

public static void main(String[] args) throws Exception {
// Validate command line options.
if (args.length != 3) {
System.err.println(
"Usage: " + HexDumpProxy.class.getSimpleName() +
" <local port> <remote host> <remote port>");
return;
}

// Parse command line options.
int localPort = Integer.parseInt(args[0]);
String remoteHost = args[1];
int remotePort = Integer.parseInt(args[2]);

System.err.println(
"Proxying *:" + localPort + " to " +
remoteHost + ':' + remotePort + " ...");

// Configure the bootstrap.
Executor executor = Executors.newCachedThreadPool();
ServerBootstrap sb = new ServerBootstrap(
new NioServerSocketChannelFactory(executor, executor));

// Set up the event pipeline factory.
ClientSocketChannelFactory cf =
new NioClientSocketChannelFactory(executor, executor);

sb.setPipelineFactory(
new HexDumpProxyPipelineFactory(cf, remoteHost, remotePort));

// Start up the server.
sb.bind(new InetSocketAddress(localPort));
}
}

//===========================

import static org.jboss.netty.channel.Channels.*;

import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;

/**
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
*/
public class HexDumpProxyPipelineFactory implements ChannelPipelineFactory {

private final ClientSocketChannelFactory cf;
private final String remoteHost;
private final int remotePort;

public HexDumpProxyPipelineFactory(
ClientSocketChannelFactory cf, String remoteHost, int remotePort) {
this.cf = cf;
this.remoteHost = remoteHost;
this.remotePort = remotePort;
}

public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline p = pipeline(); // Note the static import.
p.addLast("handler", new HexDumpProxyInboundHandler(cf, remoteHost, remotePort));
return p;
}
}

//================================

import java.net.InetSocketAddress;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;

/**
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev: 2121 $, $Date: 2010-02-02 09:38:07 +0900 (Tue, 02 Feb 2010) $
*/
public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {

private final ClientSocketChannelFactory cf;
private final String remoteHost;
private final int remotePort;

private volatile Channel outboundChannel;

public HexDumpProxyInboundHandler(
ClientSocketChannelFactory cf, String remoteHost, int remotePort) {
this.cf = cf;
this.remoteHost = remoteHost;
this.remotePort = remotePort;
}

@Override
public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
// Suspend incoming traffic until connected to the remote host.
final Channel inboundChannel = e.getChannel();
inboundChannel.setReadable(false);

// Start the connection attempt.
ClientBootstrap cb = new ClientBootstrap(cf);
cb.getPipeline().addLast("handler", new OutboundHandler(e.getChannel()));
ChannelFuture f = cb.connect(new InetSocketAddress(remoteHost, remotePort));

outboundChannel = f.getChannel();
f.addListener(new ChannelFutureListener() {
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
// Connection attempt succeeded:
// Begin to accept incoming traffic.
inboundChannel.setReadable(true);
} else {
// Close the connection if the connection attempt has failed.
inboundChannel.close();
}
}
});
}

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
ChannelBuffer msg = (ChannelBuffer) e.getMessage();
System.out.println(">>> " + ChannelBuffers.hexDump(msg));
outboundChannel.write(msg);
}

@Override
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
if (outboundChannel != null) {
closeOnFlush(outboundChannel);
}
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
throws Exception {
e.getCause().printStackTrace();
closeOnFlush(e.getChannel());
}

private static class OutboundHandler extends SimpleChannelUpstreamHandler {

private final Channel inboundChannel;

OutboundHandler(Channel inboundChannel) {
this.inboundChannel = inboundChannel;
}

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
ChannelBuffer msg = (ChannelBuffer) e.getMessage();
System.out.println("<<< " + ChannelBuffers.hexDump(msg));
inboundChannel.write(msg);
}

@Override
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
closeOnFlush(inboundChannel);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
throws Exception {
e.getCause().printStackTrace();
closeOnFlush(e.getChannel());
}
}

/**
* Closes the specified channel after all queued write requests are flushed.
*/
static void closeOnFlush(Channel ch) {
if (ch.isConnected()) {
ch.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}
}
}