`
JavaCrazyer
  • 浏览: 2986727 次
  • 性别: Icon_minigender_1
  • 来自: 河南
社区版块
存档分类

Netty使用实例

 
阅读更多

贴上两个自己写好的例子,以便备注,以下两个例子基于netty-3.5.7.Final.jar用Junit进行测试

 

第一个例子:简单的发送字符串,接收字符串“Hello, World”

 

package com.wujintao.netty;

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

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;
import org.junit.Test;

class HelloWorldServerHandler extends SimpleChannelHandler {
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
			throws Exception {
		e.getChannel().write("Hello, World");
	}

	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
		System.out.println("Unexpected exception from downstream."
				+ e.getCause());
		e.getChannel().close();
	}
}

class HelloWorldClientHandler extends SimpleChannelHandler {

	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
		String message = (String) e.getMessage();
		System.out.println(message);
		e.getChannel().close();
	}

	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
		System.out.println("Unexpected exception from downstream."
				+ e.getCause());
		e.getChannel().close();
	}
}


/**
 * Netty VS MinaNetty基于Pipeline处理,Mina基于Filter过滤
 * Netty的事件驱动模型具有更好的扩展性和易用性
 * Https,SSL,PB,RSTP,Text &Binary等协议支持
 * Netty中UDP传输有更好的支持官方测试Netty比Mina性能更好
 * @author Administrator
 *
 */
public class TestCase {

	public void testServer() {
		//初始化channel的辅助类,为具体子类提供公共数据结构
		ServerBootstrap bootstrap = new ServerBootstrap(
				new NioServerSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			public ChannelPipeline getPipeline() {
				ChannelPipeline pipeline = Channels.pipeline();
				pipeline.addLast("decoder", new StringDecoder());
				pipeline.addLast("encoder", new StringEncoder());
				pipeline.addLast("handler", new HelloWorldServerHandler());
				return pipeline;
			}
		});
		//创建服务器端channel的辅助类,接收connection请求
		bootstrap.bind(new InetSocketAddress(8080));
	}
	
	
	
	public void testClient() {
		//创建客户端channel的辅助类,发起connection请求 
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		//It means one same HelloWorldClientHandler instance is going to handle multiple Channels and consequently the data will be corrupted.
		//基于上面这个描述,必须用到ChannelPipelineFactory每次创建一个pipeline
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			public ChannelPipeline getPipeline() {
				ChannelPipeline pipeline =  Channels.pipeline();
				pipeline.addLast("decoder", new StringDecoder());
				pipeline.addLast("encoder", new StringEncoder());
				pipeline.addLast("handler", new HelloWorldClientHandler());
				return pipeline;
			}
		});
		//创建无连接传输channel的辅助类(UDP),包括client和server
		ChannelFuture future = bootstrap.connect(new InetSocketAddress(
				"localhost", 8080));
		future.getChannel().getCloseFuture().awaitUninterruptibly();
		bootstrap.releaseExternalResources();
	}
	
	
	@Test
	public void testNetty(){
		testServer();
		testClient();
	}

}

 

第二个例子,实际应用中会用到这个,发送POJO类Persons [name=周杰伦123, age=31, salary=10000.44]

package com.wujintao.netty;

import static org.jboss.netty.buffer.ChannelBuffers.dynamicBuffer;

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

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.junit.Test;

/**
 * 用POJO代替ChannelBuffer
 */

class TimeServerHandler3 extends SimpleChannelHandler {  
	  
    @Override  
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)  
            throws Exception {  
        Persons person = new Persons("周杰伦123",31,10000.44);
        ChannelFuture future = e.getChannel().write(person);  
        future.addListener(ChannelFutureListener.CLOSE);  
    }  
}  

class TimeClientHandler3 extends SimpleChannelHandler{  
	  
    @Override  
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)  
            throws Exception {  
    	Persons person = (Persons)e.getMessage();  
        System.out.println(person);  
        e.getChannel().close();  
    }  
}

/**
 * FrameDecoder and ReplayingDecoder allow you to return an object of any type.
 * 
 */
class TimeDecoder extends FrameDecoder {  
	private final ChannelBuffer buffer = dynamicBuffer();
	  
    @Override  
    protected Object decode(ChannelHandlerContext ctx, Channel channel,  
            ChannelBuffer channelBuffer) throws Exception {  
        if(channelBuffer.readableBytes()<4) {  
            return null;  
        }  
        if (channelBuffer.readable()) {
			// 读到,并写入buf
        	channelBuffer.readBytes(buffer, channelBuffer.readableBytes());
		}
        int namelength = buffer.readInt();
        String name = new String(buffer.readBytes(namelength).array(),"GBK");
        int age = buffer.readInt();
        double salary = buffer.readDouble();
        Persons person = new Persons(name,age,salary);
        return person;  
    }  
  
}  

class TimeEncoder extends SimpleChannelHandler {  
	private final ChannelBuffer buffer = dynamicBuffer();
	
    @Override  
    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e)  
            throws Exception {  
        Persons person = (Persons)e.getMessage();  
        buffer.writeInt(person.getName().getBytes("GBK").length);
        buffer.writeBytes(person.getName().getBytes("GBK"));
        buffer.writeInt(person.getAge());
        buffer.writeDouble(person.getSalary());
        Channels.write(ctx, e.getFuture(), buffer);  
    }  
}

class Persons{
	private String name;
	private int age;
	private double salary;
	
	public Persons(String name,int age,double salary){
		this.name = name;
		this.age = age;
		this.salary = salary;
	}
	
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
	public int getAge() {
		return age;
	}
	public void setAge(int age) {
		this.age = age;
	}
	public double getSalary() {
		return salary;
	}
	public void setSalary(double salary) {
		this.salary = salary;
	}

	@Override
	public String toString() {
		return "Persons [name=" + name + ", age=" + age + ", salary=" + salary
				+ "]";
	}
	
	
}

public class TestCase5 {
	public void testServer() {
		  ChannelFactory factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());  
	        ServerBootstrap bootstrap = new ServerBootstrap(factory);  
	        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {  
	              
	            public ChannelPipeline getPipeline() throws Exception {  
	                return Channels.pipeline(new TimeEncoder(), new TimeServerHandler3());  
	            }  
	        });  
	        bootstrap.setOption("child.tcpNoDelay", true);  
	        bootstrap.setOption("child.keepAlive", true);  
	          
	        bootstrap.bind(new InetSocketAddress("localhost",9999)); 
	}
	
	public void testClient(){
		//创建客户端channel的辅助类,发起connection请求 
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			public ChannelPipeline getPipeline() {
				ChannelPipeline pipeline =  Channels.pipeline();
				pipeline.addLast("decoder", new TimeDecoder());
				pipeline.addLast("encoder", new TimeEncoder());
				pipeline.addLast("handler", new TimeClientHandler3());
				return pipeline;
			}
		});
		//创建无连接传输channel的辅助类(UDP),包括client和server
		ChannelFuture future = bootstrap.connect(new InetSocketAddress(
				"localhost", 9999));
		future.getChannel().getCloseFuture().awaitUninterruptibly();
		bootstrap.releaseExternalResources();
	}

	@Test
	public void testNetty() {
			testServer();
			testClient();
	}
}
 
3
1
分享到:
评论
6 楼 似水流年依旧 2014-05-22  
把代码测试一下   不知道为什么示例2提示  java.lang.IllegalArgumentException: unsupported message type: class cn.dzy.netty.Person   这个为什么啊
5 楼 wsh525354 2013-08-30  
小笨笨->多多 写道
是楼主没写明白这个啦

private final ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();

这样就OK啦 


楼主是静态引入的:import static org.jboss.netty.buffer.ChannelBuffers.*;
4 楼 小笨笨->多多 2013-05-03  
是楼主没写明白这个啦

private final ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();

这样就OK啦 

3 楼 wcs娣 2013-04-22  
roly2000 写道
您好,看你代码,有一处不是很清楚 dynamicBuffer();这个方法怎么没看见的

原理还不太懂,但是改成这个就可以运行了
DynamicChannelBuffer dynamicChannelBuffer = new DynamicChannelBuffer(100);
    private final ChannelBuffer buffer = dynamicChannelBuffer; 
2 楼 JavaCrazyer 2013-04-01  
roly2000 写道
您好,看你代码,有一处不是很清楚 dynamicBuffer();这个方法怎么没看见的

这个是netty的jar包自带的呀
1 楼 roly2000 2013-03-29  
您好,看你代码,有一处不是很清楚 dynamicBuffer();这个方法怎么没看见的

相关推荐

Global site tag (gtag.js) - Google Analytics