netty udp server demo

之前在工作中已經用netty寫了tcp服務，感覺還不錯，就又簡單的寫了個Udp服務，防止以后工作中用到，到時就不用再到處翻了，拿來就用O(∩_∩)O~

說明：我用的是netty是3.5.3 ，截止目前3.x最新穩定版，看官網已經有4.0的alpha版了，不過4.0和3.x是不兼容的，改動比較大，等4.0穩定后再嘗試嘗試

/**
 * @author Jadic
 * @created 2012-8-10 
 */
package com.jadic;

import java.net.InetSocketAddress;

import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.DatagramChannelFactory;
import org.jboss.netty.channel.socket.nio.NioDatagramChannelFactory;

/**
 * @author Jadic
 *
 */
public class UdpServer {

    private ConnectionlessBootstrap udpBootstrap;
    
    public UdpServer(int port) {
        DatagramChannelFactory channelFactory = new NioDatagramChannelFactory();
        udpBootstrap = new ConnectionlessBootstrap(channelFactory);
        udpBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                return Channels.pipeline(new UdpEventHandler());
            }
        });
        udpBootstrap.bind(new InetSocketAddress("192.168.6.19", port));
        System.out.println("udp server started, listening on port:" + port);
    }

    public static void main(String[] args) {
        new UdpServer(6803);
    }
    
}

 

Udp數據handler，其中主要重寫下messageReceived和exceptionCaught方法就可以了

/**
 * @author Jadic
 * @created 2012-8-10 
 */
package com.jadic;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

/**
 * @author Jadic
 *
 */
public class UdpEventHandler extends SimpleChannelUpstreamHandler {
    
    private void log(Object msg) {
        System.out.println(msg);
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
            throws Exception {
        log("messageReceived");
        ChannelBuffer buffer = (ChannelBuffer)e.getMessage();
        log("recvd " + buffer.readableBytes() + " bytes [" + KKTool.channelBufferToHexStr(buffer) + "]");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
            throws Exception {
        log("exceptionCaught");
    }

}

 

有個小細節，在打印udp數據日志時，有用到ChannelBuffer.array()獲取字節內容，在tcp服務中，通過channelBuffer.array()獲取的就是真實收到的數據，而在udp服務中，則不是，跟了下源碼，tcp服務在接收數據時，是根據收到的字節數來新建一個ChannelBuffer的，而udp服務則是有根據默認接收緩存區大小新建一個ByteBuffer

如下分別是Netty Udp和Tcp服務中底層讀取數據方法的源碼，兩個類分別是NioDatagramWorker和NioWorker

 1　　  @Override
    protected boolean read(final SelectionKey key) {
        final NioDatagramChannel channel = (NioDatagramChannel) key.attachment();
        ReceiveBufferSizePredictor predictor =
            channel.getConfig().getReceiveBufferSizePredictor();
        final ChannelBufferFactory bufferFactory = channel.getConfig().getBufferFactory();
        final DatagramChannel nioChannel = (DatagramChannel) key.channel();

        // Allocating a non-direct buffer with a max udp packge size.
        // Would using a direct buffer be more efficient or would this negatively
        // effect performance, as direct buffer allocation has a higher upfront cost
        // where as a ByteBuffer is heap allocated.//此處分配ByteBuffer大小時用的是Udp接收一包最大的數據大小，默認是768
        final ByteBuffer byteBuffer = ByteBuffer.allocate(
                predictor.nextReceiveBufferSize()).order(bufferFactory.getDefaultOrder());

        boolean failure = true;
        SocketAddress remoteAddress = null;
        try {
            // Receive from the channel in a non blocking mode. We have already been notified that
            // the channel is ready to receive.
            remoteAddress = nioChannel.receive(byteBuffer);
            failure = false;
        } catch (ClosedChannelException e) {
            // Can happen, and does not need a user attention.
        } catch (Throwable t) {
            fireExceptionCaught(channel, t);
        }

        if (remoteAddress != null) {
            // Flip the buffer so that we can wrap it.
            byteBuffer.flip();

            int readBytes = byteBuffer.remaining();
            if (readBytes > 0) {
                // Update the predictor.
                predictor.previousReceiveBufferSize(readBytes);

                // Notify the interested parties about the newly arrived message.  此處將bytebuffer封裝成ChannelBuffer繼續向后觸發事件
                fireMessageReceived(
                        channel, bufferFactory.getBuffer(byteBuffer), remoteAddress);
            }
        }

Tcp的

    @Override
    protected boolean read(SelectionKey k) {
        final SocketChannel ch = (SocketChannel) k.channel();
        final NioSocketChannel channel = (NioSocketChannel) k.attachment();

        final ReceiveBufferSizePredictor predictor =
            channel.getConfig().getReceiveBufferSizePredictor();
        final int predictedRecvBufSize = predictor.nextReceiveBufferSize();

        int ret = 0;
        int readBytes = 0;
        boolean failure = true;

        ByteBuffer bb = recvBufferPool.acquire(predictedRecvBufSize);
        try {
            while ((ret = ch.read(bb)) > 0) {
                readBytes += ret;
                if (!bb.hasRemaining()) {
                    break;
                }
            }
            failure = false;
        } catch (ClosedChannelException e) {
            // Can happen, and does not need a user attention.
        } catch (Throwable t) {
            fireExceptionCaught(channel, t);
        }

        if (readBytes > 0) {
            bb.flip();

            final ChannelBufferFactory bufferFactory =
                channel.getConfig().getBufferFactory();
            final ChannelBuffer buffer = bufferFactory.getBuffer(readBytes);//這就是tcp服務中獲取到的ChannelBuffer，大小就是讀取到的字節數 readBytes
            buffer.setBytes(0, bb);
            buffer.writerIndex(readBytes);

            recvBufferPool.release(bb);

            // Update the predictor.
            predictor.previousReceiveBufferSize(readBytes);

            // Fire the event.
            fireMessageReceived(channel, buffer);
        } else {
            recvBufferPool.release(bb);
        }

        if (ret < 0 || failure) {
            k.cancel(); // Some JDK implementations run into an infinite loop without this.
            close(channel, succeededFuture(channel));
            return false;
        }

        return true;
    }