java NIO UDP通信

c86423908

浏览: 3806 次
性别:
来自: 常州

最近访客更多访客>>

风过泪落雪飘

latter

woodding2008

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

服务端代码

package com.aisuo.cn.run;

import java.io.IOException;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class AutoRunServer3 implements Runnable {

    private static final int port = 59000;

    private Charset cs = Charset.forName("utf-8");

    private static ByteBuffer sBuffer = ByteBuffer.allocate(514);

    private static ByteBuffer rBuffer = ByteBuffer.allocate(514);

    private Queue<Map<DatagramChannel, SocketAddress>> clientsQueue = new LinkedBlockingQueue<Map<DatagramChannel, SocketAddress>>();

    private static Selector selector;

    private static Queue<String> queue = new LinkedBlockingQueue<String>();

    public AutoRunServer3() {
        try {
            init();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void init() throws IOException {
        DatagramChannel serverSocketChannel = DatagramChannel.open();
        serverSocketChannel.configureBlocking(false);
        DatagramSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(port));
        selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_READ);
        System.out.println("server start on port:" + port + "；" + new Date());
    }

    private void handle(SelectionKey selectionKey) throws IOException {
        DatagramChannel client = null;
        String receiveText = null;
        //int count = 0;
        if (selectionKey.isReadable()) {
            client = (DatagramChannel) selectionKey.channel();
            rBuffer.clear();
            //count = client.read(rBuffer);
            //if (count > 0) {
            SocketAddress sa = client.receive(rBuffer);
            System.out.println(sa);
            rBuffer.flip();
            System.out.println(rBuffer.get(0));
            System.out.println(rBuffer.toString());
            byte[] ba = rBuffer.array();

            System.out.println(client.toString() + ":" + CRC16M.getBufHexStr(ba));
            //dispatch(client, receiveText);
            setClient(client, receiveText, sa);
            //client = (DatagramChannel) selectionKey.channel();
            //client.register(selector, SelectionKey.OP_READ);
            //}
        }
    }

    private void setClient(DatagramChannel client, String info, SocketAddress sa) {
        String name = "[" + sa + ":" + Integer.toHexString(client.hashCode()) + "]";
        System.out.println("插入：" + name);
        Map<DatagramChannel, SocketAddress> map = new HashMap<DatagramChannel, SocketAddress>();
        map.put(client, sa);
        clientsQueue.offer(map);
    }

    private void dispatch(String sendMsg) throws IOException, InterruptedException {
        System.out.println("sendMsg:" + sendMsg);
        Map<DatagramChannel, SocketAddress> temp = clientsQueue.poll();
        for (Map.Entry<DatagramChannel, SocketAddress> cc : temp.entrySet()) {
            DatagramChannel dc = cc.getKey();
            SocketAddress sa = cc.getValue();
            sBuffer.clear();
            sBuffer.put(sendMsg.getBytes());
            sBuffer.flip();
            //输出到通道  
            if (null != sa)
                dc.send(ByteBuffer.wrap(sendMsg.getBytes()), sa);
        }
    }

    private void putCmd() {
        /*Scanner sc = new Scanner(System.in);
        String next = sc.next();
        System.out.println("输入:" + next);*/
        for (int i = 0; i < 10; i++) {
            queue.offer("test" + i);
        }

        System.out.println("queue:" + queue.size());
    }

    public static void putCmd(String cmd) {
        queue.offer(cmd);
    }

    public static void main(String[] args) {
        AutoRunServer3 ars = new AutoRunServer3();
        //ars.putCmd();
        new Thread(ars).start();
    }

    @Override
    public void run() {
        // TODO Auto-generated method stub
        while (true) {
            try {
                selector.select();//返回值为本次触发的事件数  
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                for (SelectionKey key : selectionKeys) {
                    handle(key);
                }
                selectionKeys.clear();//清除处理过的事件  

                if (queue.size() > 0) {
                    String sendMsg = queue.poll();
                    if (null != sendMsg) {
                        //队列中存在需要发送的报文
                        dispatch(sendMsg);
                    }
                    System.out.println("Last_queue:" + queue.size());
                }
            } catch (Exception e) {
                e.printStackTrace();
                break;
            }
        }
    }

}

客户端代码：

package com.aisuo.cn.run;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;

public class ClientTest implements Runnable {

    @Override
    public void run() {
        DatagramChannel channel = null;
        Selector selector = null;
        SocketAddress sa = null;
        int udpPort = 59000;
        try {
            channel = DatagramChannel.open();
            channel.configureBlocking(false);
            sa = new InetSocketAddress("192.168.1.103", udpPort);
            selector = Selector.open();
            channel.register(selector, SelectionKey.OP_READ);
            ByteBuffer input = ByteBuffer.allocate(64);
            input.clear();
            input.put((byte)0xaa);
            input.put((byte)0x55);
            input.put((byte)0x0c);
            input.put((byte)0x00);
            input.put((byte)0x37);
            input.put((byte)0x3f);
            input.put((byte)0x01);
            input.put((byte)0x26);
            input.put((byte)0x01);
            input.put((byte)0x80);
            input.put((byte)0x01);
            input.put((byte)0x00);
            input.put((byte)0x00);
            input.put((byte)0x00);
            input.put((byte)0xfc);
            input.put((byte)0x87);

            input.flip();
            
            int count = channel.send(input, sa);
            System.out.println(count);
            ByteBuffer byteBuffer = ByteBuffer.allocate(1000);
            int eventsCount = selector.select();
            System.out.println(eventsCount);
            if (eventsCount > 0) {
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectedKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey sk = (SelectionKey) iterator.next();
                    if (sk.isReadable()) {
                        DatagramChannel datagramChannel = (DatagramChannel) sk.channel();
                        byteBuffer.clear();
                        SocketAddress isa = datagramChannel.receive(byteBuffer);
                        byteBuffer.flip();
                        System.out.println(Charset.defaultCharset().decode(byteBuffer).toString());
                        Thread.sleep(3000);//延时
                        datagramChannel.send(Charset.defaultCharset().encode("hello server"), isa);
                    }
                }
            }
        } catch (Exception e) {

        }
    }

    public static void main(String[] args) {
        new Thread(new ClientTest()).start();
    }
}

已测！！！！！

分享到：