Java NIO实例-DatagramChannel实现UDP协议传输

101 阅读 0 评论 67 点赞

我是靠谱客的博主害羞小蜜蜂，最近开发中收集的这篇文章主要介绍Java NIO实例-DatagramChannel实现UDP协议传输，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

自己的学习笔记（日后发现问题会修正）

在网上找到一些关于udp协议，学习后写了一个实例用于学习之用。

根据别人的经验，总结了以下几点内容：

TCP与UDP效率比较：

TCP协议适用于对效率要求相对低，但对准确性要求相对高的场景下，或者是有一种连接概念的场景下；而UDP协议适用于对效率要求相对高，对准确性要求相对低的场景。

TCP与UDP应用场景：

TCP可以用于网络数据库，分布式高精度计算系统的数据传输；UDP可以用于服务系统内部之间的数据传输，因为数据可能比较多，内部系统局域网内的丢包错包率又很低，即便丢包，顶多是操作无效，这种情况下，UDP经常被使用。

网上收集的资料：

TCP字节流与UDP数据报：http://network.51cto.com/art/201310/413326.htm

TCP和UDP的区别（转）：http://www.cnblogs.com/bizhu/archive/2012/05/12/2497493.html

java nio对OP_WRITE的处理解决网速慢的连接：http://blog.sina.com.cn/s/blog_783ede0301013g5n.html

测试工具：

使用clumsy工具，可以模拟网络丢包、网络延迟等恶劣环境

下载地址：http://download.csdn.net/detail/foart/8999423

服务端

package cn;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author
* @date 2015-8-7 上午11:36:25
*/
import java.nio.channels.*;
import java.nio.charset.*;
import java.net.*;
import java.io.*;
import java.util.*;
import java.nio.*;
public class DatagramChannelServerDemo {
// UDP协议服务端
private int port = 9975;
DatagramChannel channel;
private Charset charset = Charset.forName("UTF-8");
private Selector selector = null;
public DatagramChannelServerDemo() throws IOException {
try {
selector = Selector.open();
channel = DatagramChannel.open();
} catch (Exception e) {
selector = null;
channel = null;
System.out.println("超时");
}
System.out.println("服务器启动");
}
/* 编码过程 */
public ByteBuffer encode(String str) {
return charset.encode(str);
}
/* 解码过程 */
public String decode(ByteBuffer bb) {
return charset.decode(bb).toString();
}
/* 服务器服务方法 */
public void service() throws IOException {
if(channel==null || selector==null) return;
channel.configureBlocking(false);
channel.socket().bind(new InetSocketAddress(port));
// channel.write(ByteBuffer.wrap(new String("aaaa").getBytes()));
channel.register(selector, SelectionKey.OP_READ);
/** 外循环，已经发生了SelectionKey数目 */
while (selector.select() > 0) {
System.out.println("有新channel加入");
/* 得到已经被捕获了的SelectionKey的集合 */
Iterator iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = null;
try {
key = (SelectionKey) iterator.next();
iterator.remove();
if (key.isReadable()) {
reveice(key);
}
if (key.isWritable()) {
// send(key);
}
} catch (IOException e) {
e.printStackTrace();
try {
if (key != null) {
key.cancel();
key.channel().close();
}
} catch (ClosedChannelException cex) {
e.printStackTrace();
}
}
}
/* 内循环完 */
}
/* 外循环完 */
}
/*
* 接收 用receive()读IO
* 作为服务端一般不需要调用connect()，如果未调用<span style="font-family: Arial, Helvetica, sans-serif;">connect()时调</span><span style="font-family: Arial, Helvetica, sans-serif;">用read()write()读写，会报java.nio.channels</span>
* .NotYetConnectedException 只有调用connect()之后,才能使用read和write.
*/
synchronized public void reveice(SelectionKey key) throws IOException {
if (key == null)
return;
// ***用channel.receive()获取客户端消息***//
// ：接收时需要考虑字节长度
DatagramChannel sc = (DatagramChannel) key.channel();
String content = "";
// create buffer with capacity of 48 bytes
ByteBuffer buf = ByteBuffer.allocate(1024);// java里一个(utf-8)中文3字节,gbk中文占2个字节
buf.clear();
SocketAddress address = sc.receive(buf); // read into buffer. 返回客户端的地址信息
String clientAddress = address.toString().replace("/", "").split(":")[0];
String clientPost = address.toString().replace("/", "").split(":")[1];
buf.flip(); // make buffer ready for read
while (buf.hasRemaining()) {
buf.get(new byte[buf.limit()]);// read 1 byte at a time
content += new String(buf.array());
}
buf.clear(); // make buffer ready for writing
System.out.println("接收：" + content.trim());
// 第一次发；udp采用数据报模式，发送多少次，接收多少次
ByteBuffer buf2 = ByteBuffer.allocate(65507);
buf2.clear();
buf2
.put("消息推送内容 abc..UDP是一个非连接的协议，传输数据之前源端和终端不建立连接，当它想传送时就简单地去抓取来自应用程序的数据，并尽可能快地把它扔到网络上。在发送端UDP是一个非连接的协议，传输数据之前源端和终端不建立连接，当它想传送时就简单地去抓取来自应用程序的数据，并尽可能快地把它扔到网络上。在发送端UDP是一个非连接的协议，传输数据之前源端和终端不建立连接，当它想传送时就简单地去抓取来自应用程序的数据，并尽可能快地把它扔到网络上。在发送端@Q"
.getBytes());
buf2.flip();
channel.send(buf2, new InetSocketAddress(clientAddress,Integer.parseInt(clientPost))); // 将消息回送给客户端
// 第二次发
ByteBuffer buf3 = ByteBuffer.allocate(65507);
buf3.clear();
buf3.put("任务完成".getBytes());
buf3.flip();
channel.send(buf3, new InetSocketAddress(clientAddress, Integer.parseInt(clientPost))); // 将消息回送给客户端
}
int y = 0;
public void send(SelectionKey key) {
if (key == null)
return;
// ByteBuffer buff = (ByteBuffer) key.attachment();
DatagramChannel sc = (DatagramChannel) key.channel();
try {
sc.write(ByteBuffer.wrap(new String("aaaa").getBytes()));
} catch (IOException e1) {
e1.printStackTrace();
}
System.out.println("send2() " + (++y));
}
/* 发送文件 */
public void sendFile(SelectionKey key) {
if (key == null)
return;
ByteBuffer buff = (ByteBuffer) key.attachment();
SocketChannel sc = (SocketChannel) key.channel();
String data = decode(buff);
if (data.indexOf("get") == -1)
return;
String subStr = data.substring(data.indexOf(" "), data.length());
System.out.println("截取之后的字符串是 " + subStr);
FileInputStream fileInput = null;
try {
fileInput = new FileInputStream(subStr);
FileChannel fileChannel = fileInput.getChannel();
fileChannel.transferTo(0, fileChannel.size(), sc);
fileChannel.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
fileInput.close();
} catch (IOException ex) {
ex.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException {
new DatagramChannelServerDemo().service();
}
}

客户端

package cn;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author
* @date 2015-8-7 上午11:36:25
*/
import java.nio.channels.*;
import java.nio.charset.*;
import java.net.*;
import java.io.*;
import java.util.*;
import java.nio.*;
public class DatagramChannelClientDemo {
// UDP协议客户端
private String serverIp = "127.0.0.1";
private int port = 9975;
// private ServerSocketChannel serverSocketChannel;
DatagramChannel channel;
private Charset charset = Charset.forName("UTF-8");
private Selector selector = null;
public DatagramChannelClientDemo() throws IOException {
try {
selector = Selector.open();
channel = DatagramChannel.open();
} catch (Exception e) {
selector = null;
channel = null;
System.out.println("超时");
}
System.out.println("客户器启动");
}
/* 编码过程 */
public ByteBuffer encode(String str) {
return charset.encode(str);
}
/* 解码过程 */
public String decode(ByteBuffer bb) {
return charset.decode(bb).toString();
}
/* 服务器服务方法 */
public void service() throws IOException {
if(channel==null || selector==null) return;
channel.configureBlocking(false);
channel.connect(new InetSocketAddress(serverIp, port));// 连接服务端
channel.write(ByteBuffer.wrap(new String("客户端请求获取消息").getBytes()));
channel.register(selector, SelectionKey.OP_READ);
/** 外循环，已经发生了SelectionKey数目 */
while (selector.select() > 0) {
/* 得到已经被捕获了的SelectionKey的集合 */
Iterator iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = null;
try {
key = (SelectionKey) iterator.next();
iterator.remove();
if (key.isReadable()) {
reveice(key);
}
if (key.isWritable()) {
// send(key);
}
} catch (IOException e) {
e.printStackTrace();
try {
if (key != null) {
key.cancel();
key.channel().close();
}
} catch (ClosedChannelException cex) {
e.printStackTrace();
}
}
}
/* 内循环完 */
}
/* 外循环完 */
}
//	/*
//
* 接收	用read()读IO
//
*
*/
//	synchronized public void reveice2(SelectionKey key) throws IOException {
//
if (key == null)
//
return;
//
// ***用channel.read()获取消息***//
//
// ：接收时需要考虑字节长度
//
DatagramChannel sc = (DatagramChannel) key.channel();
//
String content = "";
//
// create buffer with capacity of 48 bytes
//
ByteBuffer buf = ByteBuffer.allocate(3);// java里一个(utf-8)中文3字节,gbk中文占2个字节
//
int bytesRead = sc.read(buf); //read into buffer.
//
//
while (bytesRead >0) {
//
buf.flip();
//make buffer ready for read
//
while(buf.hasRemaining()){
//
buf.get(new byte[buf.limit()]); // read 1 byte at a time
//
content += new String(buf.array());
//
}
//
buf.clear(); //make buffer ready for writing
//
bytesRead = sc.read(buf);
//
}
//
System.out.println("接收：" + content);
//	}
/* 接收 */
synchronized public void reveice(SelectionKey key) throws IOException {
String threadName = Thread.currentThread().getName();
if (key == null)
return;
try {
// ***用channel.receive()获取消息***//
// ：接收时需要考虑字节长度
DatagramChannel sc = (DatagramChannel) key.channel();
String content = "";
//第一次接；udp采用数据报模式，发送多少次，接收多少次
ByteBuffer buf = ByteBuffer.allocate(65507);// java里一个(utf-8)中文3字节,gbk中文占2个字节
buf.clear();
SocketAddress address = sc.receive(buf); // read into buffer.
String clientAddress = address.toString().replace("/", "").split(":")[0];
String clientPost = address.toString().replace("/", "").split(":")[1];
System.out.println(threadName + "t" + address.toString());
buf.flip(); // make buffer ready for read
while (buf.hasRemaining()) {
buf.get(new byte[buf.limit()]);// read 1 byte at a time
byte[] tmp = buf.array();
content += new String(tmp);
}
buf.clear(); // make buffer ready for writing次
System.out.println(threadName + "接收：" + content.trim());
//第二次接
content = "";
ByteBuffer buf2 = ByteBuffer.allocate(65507);// java里一个(utf-8)中文3字节,gbk中文占2个字节
buf2.clear();
SocketAddress address2 = sc.receive(buf2); // read into buffer.
buf2.flip(); // make buffer ready for read
while (buf2.hasRemaining()) {
buf2.get(new byte[buf2.limit()]);// read 1 byte at a time
byte[] tmp = buf2.array();
content += new String(tmp);
}
buf2.clear(); // make buffer ready for writing次
System.out.println(threadName + "接收2：" + content.trim());
} catch (PortUnreachableException ex) {
System.out.println(threadName + "服务端端口未找到!");
}
send(2);
}
boolean flag = false;
public void send(int i) {
if (flag)
return;
try {
// channel.write(ByteBuffer.wrap(new String("客户端请求获取消息(第"+i+"次)").getBytes()));
// channel.register(selector, SelectionKey.OP_READ );
ByteBuffer buf2 = ByteBuffer.allocate(48);
buf2.clear();
buf2.put(("客户端请求获取消息(第" + i + "次)").getBytes());
buf2.flip();
channel.write(buf2);
channel.register(selector, SelectionKey.OP_READ );
//
int bytesSent = channel.send(buf2, new InetSocketAddress(serverIp,port)); // 将消息回送给服务端
} catch (IOException e) {
e.printStackTrace();
}
flag = true;
}
int y = 0;
public void send(SelectionKey key) {
if (key == null)
return;
// ByteBuffer buff = (ByteBuffer) key.attachment();
DatagramChannel sc = (DatagramChannel) key.channel();
try {
sc.write(ByteBuffer.wrap(new String("aaaa").getBytes()));
} catch (IOException e1) {
e1.printStackTrace();
}
System.out.println("send2() " + (++y));
}
/* 发送文件 */
public void sendFile(SelectionKey key) {
if (key == null)
return;
ByteBuffer buff = (ByteBuffer) key.attachment();
SocketChannel sc = (SocketChannel) key.channel();
String data = decode(buff);
if (data.indexOf("get") == -1)
return;
String subStr = data.substring(data.indexOf(" "), data.length());
System.out.println("截取之后的字符串是 " + subStr);
FileInputStream fileInput = null;
try {
fileInput = new FileInputStream(subStr);
FileChannel fileChannel = fileInput.getChannel();
fileChannel.transferTo(0, fileChannel.size(), sc);
fileChannel.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
fileInput.close();
} catch (IOException ex) {
ex.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException {
new Thread(new Runnable() {
public void run() {
try {
new DatagramChannelClientDemo().service();
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
//
new Thread(new Runnable() {
//
public void run() {
//
try {
//
new DatagramChannelClientDemo().service();
//
} catch (IOException e) {
//
e.printStackTrace();
//
}
//
}
//
}).start();
}
}