netty的timeout

客户端

 

 

Java代码  

1. package com.mchz.netty.test.client;  
2.   
3. import java.net.InetSocketAddress;  
4. import java.util.concurrent.Executors;  
5.   
6. import org.jboss.netty.bootstrap.ClientBootstrap;  
7. import org.jboss.netty.channel.ChannelFuture;  
8. import org.jboss.netty.channel.ChannelPipeline;  
9. import org.jboss.netty.channel.ChannelPipelineFactory;  
10. import org.jboss.netty.channel.Channels;  
11. import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;  
12.   
13. /** 
14.  * Sends one message when a connection is open and echoes back any received data 
15.  * to the server. Simply put, the echo client initiates the ping-pong traffic 
16.  * between the echo client and server by sending the first message to the 
17.  * server. 
18.  */  
19. public class EchoClient extends Thread {  
20.   
21.     private final String host;  
22.     private final int port;  
23.     private final int firstMessageSize;  
24.   
25.     private Integer recyle = 5;  
26.   
27.     public EchoClient(String host, int port, int firstMessageSize,  
28.             String threadName) {  
29.         this.host = host;  
30.         this.port = port;  
31.         this.firstMessageSize = firstMessageSize;  
32.         System.out.println("current thread name is ====" + threadName);  
33.         this.start();  
34.     }  
35.   
36.     public void run() {  
37.         ClientBootstrap bootstrap = new ClientBootstrap(  
38.                 new NioClientSocketChannelFactory(  
39.                         Executors.newCachedThreadPool(),  
40.                         Executors.newCachedThreadPool()));  
41.         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {  
42.             public ChannelPipeline getPipeline() throws Exception {  
43.                 return Channels.pipeline(new EchoClientHandler(  
44.                         firstMessageSize, recyle));  
45.             }  
46.         });  
47.   
48.         ChannelFuture future = bootstrap.connect(new InetSocketAddress(host,  
49.                 port));  
50.         future.getChannel().getCloseFuture().awaitUninterruptibly();  
51.         bootstrap.setOption("child.tcpNoDelay", true);  
52.         bootstrap.setOption("child.keepAlive", true);  
53.         bootstrap.releaseExternalResources();  
54.     }  
55.   
56.     public static void main(String[] args) throws Exception {  
57.         int i = 1;  
58.         while (true) {  
59.             i++;  
60. //          new EchoClient("172.16.4.123", 8080, 256, "thread=" + i);  
61.             new EchoClient("127.0.0.1", 8080, 256, "thread=" + i);  
62.             if (i > 3) {  
63.                 break;  
64.             }  
65.         }  
66.   
67.         Thread.sleep(1000 * 200);  
68.         System.out.println("end....");  
69.     }  
70. }  

 

 

Java代码  

1. package com.mchz.netty.test.client;  
2.   
3. import java.util.concurrent.atomic.AtomicLong;  
4. import java.util.logging.Level;  
5. import java.util.logging.Logger;  
6.   
7. import org.jboss.netty.buffer.ChannelBuffer;  
8. import org.jboss.netty.buffer.ChannelBuffers;  
9. import org.jboss.netty.channel.ChannelHandlerContext;  
10. import org.jboss.netty.channel.ChannelStateEvent;  
11. import org.jboss.netty.channel.ExceptionEvent;  
12. import org.jboss.netty.channel.SimpleChannelUpstreamHandler;  
13.   
14.   
15. /** 
16.  * Handler implementation for the echo client. It initiates the ping-pong 
17.  * traffic between the echo client and server by sending the first message to 
18.  * the server. 
19.  */  
20. public class EchoClientHandler extends SimpleChannelUpstreamHandler {  
21.   
22.     private static final Logger logger = Logger  
23.             .getLogger(EchoClientHandler.class.getName());  
24.     private Integer recyle=5;  
25.     private final ChannelBuffer firstMessage;  
26.     private final AtomicLong transferredBytes = new AtomicLong();  
27.   
28.     /** 
29.      * Creates a client-side handler. 
30.      */  
31.     public EchoClientHandler(int firstMessageSize,Integer recyle) {  
32.           
33.         this.recyle=recyle;  
34.         if (firstMessageSize <= 0) {  
35.             throw new IllegalArgumentException("firstMessageSize: "  
36.                     + firstMessageSize);  
37.         }  
38.         firstMessage = ChannelBuffers.buffer(firstMessageSize);  
39.         for (int i = 0; i < firstMessage.capacity(); i++) {  
40.             firstMessage.writeByte((byte) i);  
41.         }  
42.     }  
43.   
44.     public long getTransferredBytes() {  
45.         return transferredBytes.get();  
46.     }  
47.   
48.     @Override  
49.     public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {  
50.           
51.         for (int i = 0; i < recyle; i++) {  
52.             try {  
53.                 System.out.println("send a message to server ...");  
54.                 e.getChannel().write(firstMessage);  
55.                 Thread.sleep(5000);  
56.             } catch (InterruptedException e1) {  
57.                 e1.printStackTrace();  
58.             }  
59.         }  
60.     }  
61.   
62.   
63.     @Override  
64.     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {  
65.         // Close the connection when an exception is raised.  
66.         System.out.println("close the connection when an exception is raised");  
67.         logger.log(Level.WARNING, "Unexpected exception from downstream.",  
68.                 e.getCause());  
69.         e.getChannel().close();  
70.     }  
71. }  

 

服务端

 

    package com.mchz.netty.test.server;

Java代码  

1. import java.net.InetSocketAddress;  
2. import java.util.concurrent.Executors;  
3.   
4. import org.jboss.netty.bootstrap.ServerBootstrap;  
5. import org.jboss.netty.channel.ChannelPipelineFactory;  
6. import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;  
7.   
8. /** 
9.  * Echoes back any received data from a client. 
10.  */  
11. public class EchoServer {  
12.   
13.     private final int port;  
14.   
15.     public EchoServer(int port) {  
16.         this.port = port;  
17.     }  
18.   
19.     public void run() {  
20.         // Configure the server.  
21.         ServerBootstrap bootstrap = new ServerBootstrap(  
22.                 new NioServerSocketChannelFactory(  
23.                         Executors.newCachedThreadPool(),  
24.                         Executors.newCachedThreadPool()));  
25.   
26.         ChannelPipelineFactory pipelineFactory = new MyPipelineFactory(  
27.                 new EchoServerHandler());  
28.         bootstrap.setPipelineFactory(pipelineFactory);  
29. //      bootstrap.setOption("allIdleTime", "10");  
30.           
31.         bootstrap.bind(new InetSocketAddress(port));  
32.     }  
33.   
34.     public static void main(String[] args) throws Exception {  
35.         int port;  
36.         if (args.length > 0) {  
37.             port = Integer.parseInt(args[0]);  
38.         } else {  
39.             port = 8080;  
40.         }  
41.         new EchoServer(port).run();  
42.     }  
43. }  

 package com.mchz.netty.test.server;

Java代码  

1. import java.util.concurrent.atomic.AtomicLong;  
2. import java.util.logging.Level;  
3. import java.util.logging.Logger;  
4.   
5. import org.jboss.netty.buffer.ChannelBuffer;  
6. import org.jboss.netty.channel.ChannelHandlerContext;  
7. import org.jboss.netty.channel.ChannelStateEvent;  
8. import org.jboss.netty.channel.ExceptionEvent;  
9. import org.jboss.netty.channel.MessageEvent;  
10. import org.jboss.netty.handler.timeout.IdleState;  
11. import org.jboss.netty.handler.timeout.IdleStateAwareChannelHandler;  
12. import org.jboss.netty.handler.timeout.IdleStateEvent;  
13. /** 
14.  * Handler implementation for the echo server. 
15.  */  
16. public class EchoServerHandler extends IdleStateAwareChannelHandler  {  
17.   
18.     private static final Logger logger = Logger  
19.             .getLogger(EchoServerHandler.class.getName());  
20.     private final AtomicLong transferredBytes = new AtomicLong();  
21.   
22.     public long getTransferredBytes() {  
23.         return transferredBytes.get();  
24.     }  
25.   
26.     @Override  
27.     public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)  
28.             throws Exception {  
29.         System.out.println("server has been connected");  
30.         super.channelConnected(ctx, e);  
31.           
32.     }  
33.   
34.     @Override  
35.     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {  
36.         // Send back the received message to the remote peer.  
37.         transferredBytes.addAndGet(((ChannelBuffer) e.getMessage())  
38.                 .readableBytes());  
39.         System.out  
40.                 .println("I an server ,I received a message,and I will received a message after 5 mill later");  
41. //      try {  
42. //          Thread.sleep(5000);  
43. //      } catch (InterruptedException e1) {  
44. //          e1.printStackTrace();  
45. //      }  
46. //      e.getChannel().write(e.getMessage());  
47.     }  
48.   
49.     @Override  
50.     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {  
51.         // Close the connection when an exception is raised.  
52.         System.out.println(" Close the connection when an exception is raised"+e.getCause().getMessage());  
53.         logger.log(Level.WARNING, "Unexpected exception from downstream.",  
54.                 e.getCause());  
55.         e.getChannel().close();  
56.     }  
57.       
58.     @Override  
59.     public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e)  
60.             throws Exception {  
61.           
62. //      super.channelIdle(ctx, e);  
63.         if( e.getState() == IdleState.ALL_IDLE){    
64. ////            e.getChannel().write("str123".getBytes());    
65.             super.channelIdle(ctx, e);        
66.          }    
67.     }  
68.       
69.       
70. }  

 package com.mchz.netty.test.server;

Java代码  

1. import org.jboss.netty.channel.ChannelHandler;  
2. import org.jboss.netty.channel.ChannelPipeline;  
3. import org.jboss.netty.channel.ChannelPipelineFactory;  
4. import org.jboss.netty.channel.Channels;  
5. import org.jboss.netty.handler.timeout.ReadTimeoutHandler;  
6. import org.jboss.netty.util.HashedWheelTimer;  
7. import org.jboss.netty.util.Timer;  
8.   
9. public class MyPipelineFactory implements ChannelPipelineFactory {  
10.     private ChannelHandler serverHandler;  
11.   
12.     public MyPipelineFactory(ChannelHandler serverHander) {  
13.         this.serverHandler = serverHander;  
14.     }  
15.   
16.     public ChannelPipeline getPipeline() throws Exception {  
17.         ChannelPipeline pipeline = Channels.pipeline();  
18.         Timer timer = new HashedWheelTimer();  
19.         pipeline.addLast("timeout", new ReadTimeoutHandler(timer, 10));  
20.         pipeline.addLast("idleHandler", serverHandler);  
21.         return pipeline;  
22.     }  
23.   
24. }  

• netty-3.4.5.Final.jar (950.7 KB)
• 下载次数: 6

• Netty_3.1中文用户手册.zip (119.2 KB)
• 下载次数: 16

• netty-test.zip (1.7 MB)
• 下载次数: 21