转载自http://blog.csdn.net/kobejayandy/article/details/11493717
先啰嗦两句,如果你还不知道Netty是做什么的能做什么。那可以先简单的搜索了解一下。我只能说Netty是一个NIO的框架,可以用于开发分布式的Java程序。具体能做什么,各位可以尽量发挥想象。技术,是服务于人而不是局限住人的。
如果你已经万事具备,那么我们先从一段代码开始。程序员们习惯的上手第一步,自然是"Hello world",不过Netty官网的例子却偏偏抛弃了"Hello world"。那我们就自己写一个最简单的"Hello world"的例子,作为上手。
- /**
- * Netty 服务端代码
- *
- * @author lihzh
- * @alia OneCoder
- * @blog http://www.coderli.com
- */
- public class HelloServer {
- public static void main(String args[]) {
- // Server服务启动器
- ServerBootstrap bootstrap = new ServerBootstrap(
- new NioServerSocketChannelFactory(
- Executors.newCachedThreadPool(),
- Executors.newCachedThreadPool()));
- // 设置一个处理客户端消息和各种消息事件的类(Handler)
- bootstrap
- .setPipelineFactory(new ChannelPipelineFactory() {
- @Override
- public ChannelPipeline getPipeline()
- throws Exception {
- return Channels
- .pipeline(new HelloServerHandler());
- }
- });
- // 开放8000端口供客户端访问。
- bootstrap.bind(new InetSocketAddress(8000));
- }
- private static class HelloServerHandler extends
- SimpleChannelHandler {
- /**
- * 当有客户端绑定到服务端的时候触发,打印"Hello world, I'm server."
- *
- * @alia OneCoder
- * @author lihzh
- */
- @Override
- public void channelConnected(
- ChannelHandlerContext ctx,
- ChannelStateEvent e) {
- System.out.println("Hello world, I'm server.");
- }
- }
- }
- /**
- * Netty 客户端代码
- *
- * @author lihzh
- * @alia OneCoder
- * @blog http://www.coderli.com
- */
- public class HelloClient {
- public static void main(String args[]) {
- // Client服务启动器
- ClientBootstrap bootstrap = new ClientBootstrap(
- new NioClientSocketChannelFactory(
- Executors.newCachedThreadPool(),
- Executors.newCachedThreadPool()));
- // 设置一个处理服务端消息和各种消息事件的类(Handler)
- bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
- @Override
- public ChannelPipeline getPipeline() throws Exception {
- return Channels.pipeline(new HelloClientHandler());
- }
- });
- // 连接到本地的8000端口的服务端
- bootstrap.connect(new InetSocketAddress(
- "127.0.0.1", 8000));
- }
- private static class HelloClientHandler extends SimpleChannelHandler {
- /**
- * 当绑定到服务端的时候触发,打印"Hello world, I'm client."
- *
- * @alia OneCoder
- * @author lihzh
- */
- @Override
- public void channelConnected(ChannelHandlerContext ctx,
- ChannelStateEvent e) {
- System.out.println("Hello world, I'm client.");
- }
- }
- }
既然是分布式的,自然要分多个服务。Netty中,需要区分Server和Client服务。所有的Client都是绑定在Server上的,他们之间是不能通过Netty直接通信的。(自己采用的其他手段,不包括在内。)。白话一下这个通信过程,Server端开放端口,供Client连接,Client发起请求,连接到Server指定的端口,完成绑定。随后便可自由通信。其实就是普通Socket连接通信的过程。
Netty框架是基于事件机制的,简单说,就是发生什么事,就找相关处理方法。就跟着火了找119,抢劫了找110一个道理。所以,这里,我们处理的是当客户端和服务端完成连接以后的这个事件。什么时候完成的连接,Netty知道,他告诉我了,我就负责处理。这就是框架的作用。Netty,提供的事件还有很多,以后会慢慢的接触和介绍。
你应该已经可以上手了:)
"Hello World"的代码固然简单,不过其中的几个重要概念(类)和 Netty的工作原理还是需要简单明确一下,至少知道其是负责什。方便自己以后更灵活的使用和扩展。
- ChannelEvent
- ChannelPipeline
- ChannelHandler
- Channel
这些基本的概念,你懂了吧。
说了这么多废话,才提到对象的传输,不知道您是不是已经不耐烦了。一个系统内部的消息传递,没有对象传递是不太现实的。下面就来说说,怎么传递对象。
如果,您看过前面的介绍,如果您善于专注本质,勤于思考。您应该也会想到,我们说过,Netty的消息传递都是基于流,通过ChannelBuffer传递的,那么自然,Object也需要转换成ChannelBuffer来传递。好在Netty本身已经给我们写好了这样的转换工具。 ObjectEncoder和ObjectDecoder。
工具怎么用?再一次说说所谓的本质,我们之前也说过,Netty给我们处理自己业务的空间是在灵活的可子定义的Handler上的,也就是说,如果我们自己去做这个转换工作,那么也应该在Handler里去做。而Netty,提供给我们的ObjectEncoder和Decoder也恰恰是一组 Handler。于是,修改Server和Client的启动代码:
01 02 03 04 05 06 07 08 09 10 | // 设置一个处理客户端消息和各种消息事件的类(Handler) bootstrap.setPipelineFactory( new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() throws Exception { return Channels.pipeline( new ObjectDecoder(ClassResolvers.cacheDisabled( this .getClass().getClassLoader())), new ObjectServerHandler()); } }); |
1 2 3 4 5 6 7 8 | // 设置一个处理服务端消息和各种消息事件的类(Handler) bootstrap.setPipelineFactory( new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() throws Exception { return Channels.pipeline( new ObjectEncoder(), new ObjectClientHandler()); } }); |
要传递对象,自然要有一个被传递模型,一个简单的Pojo,当然,实现序列化接口是必须的。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | /** * @author lihzh * @alia OneCoder * @bloghttp://www.coderli.com */ public class Command implements Serializable { private static final long serialVersionUID = 7590999461767050471L; private String actionName; public String getActionName() { return actionName; } public void setActionName(String actionName) { this .actionName = actionName; } } |
服务端和客户端里,我们自定义的Handler实现如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | /** * 对象传递服务端代码 * * @author lihzh * @alia OneCoder * @bloghttp://www.coderli.com */ public class ObjectServerHandler extends SimpleChannelHandler { /** * 当接受到消息的时候触发 */ @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { Command command = (Command) e.getMessage(); // 打印看看是不是我们刚才传过来的那个 System.out.println(command.getActionName()); } } |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 | /** * 对象传递,客户端代码 * * @author lihzh * @alia OneCoder * @bloghttp://www.coderli.com */ public class ObjectClientHandler extends SimpleChannelHandler { /** * 当绑定到服务端的时候触发,给服务端发消息。 * * @author lihzh * @alia OneCoder */ @Override public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) { // 向服务端发送Object信息 sendObject(e.getChannel()); } /** * 发送Object * * @param channel * @author lihzh * @alia OneCoder */ private void sendObject(Channel channel) { Command command = new Command(); command.setActionName( "Hello action." ); channel.write(command); } } |
启动后,服务端正常打印结果:Hello action.
简单梳理一下思路:
通过Netty传递,都需要基于流,以ChannelBuffer的形式传递。所以,Object -> ChannelBuffer.
Netty提供了转换工具,需要我们配置到Handler。
样例从客户端 -> 服务端,单向发消息,所以在客户端配置了编码,服务端解码。如果双向收发,则需要全部配置Encoder和Decoder。
这里需要注意,注册到Server的Handler是有顺序的,如果你颠倒一下注册顺序:
1 2 3 4 5 6 7 8 9 | bootstrap.setPipelineFactory( new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() throws Exception { return Channels.pipeline( new ObjectServerHandler(), new ObjectDecoder(ClassResolvers.cacheDisabled( this .getClass().getClassLoader())) ); } }); |
结果就是,会先进入我们自己的业务,再进行解码。这自然是不行的,会强转失败。至此,你应该会用Netty传递对象了吧。
有一段事件没有更新文章了,各种原因都有吧。搬家的琐事,搬家后的安逸呵呵。不过,OneCoder明白,绝不能放松。对于Netty的学习,也该稍微深入一点了。
01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | private void openSelector() { try { selector = Selector.open(); } catch (Throwable t) { throw new ChannelException( "Failed to create a selector." , t); } // Start the worker thread with the new Selector. boolean success = false ; try { DeadLockProofWorker.start(executor, new ThreadRenamingRunnable( this , "New I/O worker #" + id)); success = true ; } finally { if (!success) { // Release the Selector if the execution fails. try { selector.close(); } catch (Throwable t) { logger.warn( "Failed to close a selector." , t); } selector = null ; // The method will return to the caller at this point. } } assert selector != null && selector.isOpen(); } |
至此,会分线程启动AbstractNioWorker中run逻辑。同样是循环处理任务队列。
1 2 3 4 | processRegisterTaskQueue(); processEventQueue(); processWriteTaskQueue(); processSelectedKeys(selector.selectedKeys()); |