這幾天需要實(shí)現(xiàn)一個(gè)底層基于UDP的協(xié)議，該協(xié)議底層使用UDP傳輸?shù)蔷哂袚砣刂啤⒊瑫r(shí)重發(fā)、數(shù)據(jù)確認(rèn)等功能又比TCP簡(jiǎn)單 （RUDP，Reliable UDP）。在實(shí)現(xiàn)協(xié)議底層的UDP服務(wù)時(shí)準(zhǔn)備使用Java的NIO，在網(wǎng)上查資料都是以TCP為例講的，于是自己研究了一下基于UDP的NIO。
NIO的思路是基于多路選擇的，即由原來(lái)的每個(gè)連接都由一個(gè)線程來(lái)等待消息，改為每個(gè)連接都在選擇器上注冊(cè)，由選擇器來(lái)等待。當(dāng)然NIO引入了很多新的概念，如Channel，Buffer、Charset、Selector等，使得編程更簡(jiǎn)潔、更面向?qū)ο蠡?/font>
下面貼出用NIO API改造成UDP示例代碼，注意其中使用Charset來(lái)編碼解碼的過(guò)程（當(dāng)然Charset還支持很多其他編碼不僅局限于默認(rèn)編碼）以及Buffer的使用。
package
sinpo.usagedemo;
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.util.Iterator;
import
java.util.Set;
/**
*
@author
徐辛波(sinpo.xu@hotmail.com) Oct 19, 2008
*/
public class
UDPServer
extends
Thread
{
public
void
run
() {
Selector selector =
null
;
try
{
DatagramChannel channel = DatagramChannel.open
()
;
DatagramSocket socket = channel.socket
()
;
channel.configureBlocking
(
false
)
;
socket.bind
(
new
InetSocketAddress
(
5057
))
;
selector = Selector.open
()
;
channel.register
(
selector, SelectionKey.OP_READ
)
;
}
catch
(
Exception e
) {
e.printStackTrace
()
;
}
ByteBuffer byteBuffer = ByteBuffer.allocate
(
65536
)
;
while
(
true
) {
try
{
int
eventsCount = selector.select
()
;
if
(
eventsCount >
0
) {
Set selectedKeys = selector.selectedKeys
()
;
Iterator iterator = selectedKeys.iterator
()
;
while
(
iterator.hasNext
()) {
SelectionKey sk =
(
SelectionKey
)
iterator.next
()
;
iterator.remove
()
;
if
(
sk.isReadable
()) {
DatagramChannel datagramChannel =
(
DatagramChannel
)
sk
.channel
()
;
SocketAddress sa = datagramChannel
.receive
(
byteBuffer
)
;
byteBuffer.flip
()
;
// 測(cè)試：通過(guò)將收到的ByteBuffer首先通過(guò)缺省的編碼解碼成CharBuffer 再輸出
CharBuffer charBuffer = Charset.defaultCharset
()
.decode
(
byteBuffer
)
;
System.out.println
(
"receive message:"
+ charBuffer.toString
())
;
byteBuffer.clear
()
;
String echo =
"This is the reply message from 服務(wù)器。"
;
ByteBuffer buffer = Charset.defaultCharset
()
.encode
(
echo
)
;
datagramChannel.write
(
buffer
)
;
}
}
}
}
catch
(
Exception e
) {
e.printStackTrace
()
;
}
}
}
public static
void
main
(
String
[]
args
) {
new
UDPServer
()
.start
()
;
}
}
|
Client
package
sinpo.usagedemo;
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;
/**
*
@author
徐辛波(sinpo.xu@hotmail.com)
* Oct 19, 2008
*/
public class
UDPClient
extends
Thread
{
public
void
run
() {
DatagramChannel channel =
null
;
Selector selector =
null
;
try
{
channel = DatagramChannel.open
()
;
channel.configureBlocking
(
false
)
;
SocketAddress sa =
new
InetSocketAddress
(
"localhost"
,
5057
)
;
channel.connect
(
sa
)
;
}
catch
(
Exception e
) {
e.printStackTrace
()
;
}
try
{
selector = Selector.open
()
;
channel.register
(
selector, SelectionKey.OP_READ
)
;
channel.write
(
Charset.defaultCharset
()
.encode
(
"Tell me your time"
))
;
}
catch
(
Exception e
) {
e.printStackTrace
()
;
}
ByteBuffer byteBuffer = ByteBuffer.allocate
(
100
)
;
while
(
true
) {
try
{
int
eventsCount = selector.select
()
;
if
(
eventsCount >
0
) {
Set selectedKeys = selector.selectedKeys
()
;
Iterator iterator = selectedKeys.iterator
()
;
while
(
iterator.hasNext
()) {
SelectionKey sk =
(
SelectionKey
)
iterator.next
()
;
iterator.remove
()
;
if
(
sk.isReadable
()) {
DatagramChannel datagramChannel =
(
DatagramChannel
)
sk
.channel
()
;
datagramChannel.read
(
byteBuffer
)
;
byteBuffer.flip
()
;
//TODO 將報(bào)文轉(zhuǎn)化為RUDP消息并調(diào)用RUDP協(xié)議處理器來(lái)處理
System.out.println
(
Charset.defaultCharset
()
.decode
(
byteBuffer
)
.toString
())
;
byteBuffer.clear
()
;
datagramChannel.write
(
Charset.defaultCharset
()
.encode
(
"Tell me your time"
))
;
}
}
}
}
catch
(
Exception e
) {
e.printStackTrace
()
;
}
}
}
}
|