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();
}
}

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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
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();
}
}

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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
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();
}
}