一、UDP Server
項(xiàng)目的需要，需要利用java實(shí)現(xiàn)一個(gè)udp server，主要的功能是偵聽(tīng)來(lái)自客戶端的udp請(qǐng)求，客戶請(qǐng)求可能是大并發(fā)量的，對(duì)于每個(gè)請(qǐng)求Server端的處理很簡(jiǎn)單，處理每個(gè)請(qǐng)求的時(shí)間大約在1ms左右，但是Server端需要維護(hù)一個(gè)對(duì)立于請(qǐng)求的全局變量Cache，項(xiàng)目本身已經(jīng)采用Mina架構(gòu)（http://mina.apache.org/），我要開(kāi)發(fā)的Server作為整個(gè)項(xiàng)目的一個(gè)模塊，由于之前沒(méi)有開(kāi)發(fā)UDP Server，受TCP Server的影響，很自然的想利用多線程來(lái)實(shí)現(xiàn)，對(duì)于每個(gè)客戶請(qǐng)求，新建一個(gè)線程來(lái)處理相應(yīng)的邏輯，在實(shí)現(xiàn)的過(guò)程中，利用Mina的Thread Model，實(shí)現(xiàn)了一個(gè)多線程的UDP Server，但是由于要維護(hù)一個(gè)全局Cache，需要在各線程之間同步，加之處理請(qǐng)求的時(shí)間很短，很快就發(fā)現(xiàn)在此利用多線程，并不能提高性能，于是決定采用單線程來(lái)實(shí)現(xiàn)，在動(dòng)手之前，還是發(fā)現(xiàn)有兩種方案來(lái)實(shí)現(xiàn)單線程UDP Server：
1） 采用JDK的DatagramSocket和DatagramPacket來(lái)實(shí)現(xiàn)
public class UDPServerUseJDK extends Thread{
/**
* Constructor
* @param port port used to listen incoming connection
* @throws SocketException error to consturct a DatagramSocket with this port
*/
public MediatorServerUseJDK(int port) throws SocketException{
this("MediatorServerThread", port);
}
/**
* Constructor
* @param name the thread name
* @param port port used to listen incoming connection
* @throws SocketException error to consturct a DatagramSocket with this port
*/
public MediatorServerUseJDK(String name, int port) throws SocketException{
super(name);
socket = new DatagramSocket(port);
System.out.println("Mediator server started on JDK model...");
System.out.println("Socket buffer size: " + socket.getReceiveBufferSize());
}
public void run(){
long startTime = 0;
while(true){
try {
buf = new byte[1024];
// receive request
packet = new DatagramPacket(buf, buf.length);
socket.receive(packet);
.........
}catch (IOException e) {
.........
}
}
}
｝
2） 采用Mina的DatagramAcceptor來(lái)實(shí)現(xiàn)，在創(chuàng)建Exector的時(shí)候，只傳遞單個(gè)線程
public class MediatorServerUseMina {
private DatagramAcceptor acceptor;
public MediatorServerUseMina() {
}
public void startListener(InetSocketAddress address, IoHandler handler) {
// create an acceptor with a single thread
this.acceptor = new DatagramAcceptor(Executors.newSingleThreadExecutor());
// configure the thread models
DatagramAcceptorConfig acceptorConfig = acceptor.getDefaultConfig();
acceptorConfig.setThreadModel(ThreadModel.MANUAL);
// set the acceptor to reuse the address
acceptorConfig.getSessionConfig().setReuseAddress(true);
// add io filters
DefaultIoFilterChainBuilder filterChainBuilder = acceptor.getFilterChain();
// add CPU-bound job first,
filterChainBuilder.addLast("codec", new ProtocolCodecFilter(new StringCodecFactory()));
try {
// bind
acceptor.bind(address, handler);
System.out.println("Mediator Server started on mina model...");
System.out.println("Socket buffer size: " + acceptorConfig.getSessionConfig().getReceiveBufferSize());
} catch (IOException e) {
System.err.println("Error starting component listener on port(UDP) " + address.getPort() + ": "
+ e.getMessage());
}
}
}
二、Performance Test
為了測(cè)試兩個(gè)Server的性能，寫(xiě)了個(gè)簡(jiǎn)單的測(cè)試客戶端
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
/**
* @author Herry Hong
*
*/
public class PerformanceTest {
/** Number of threads to be created */
static final int THREADS = 100;
/** Packets to be sent per thread */
static final int PACKETS = 500;
/** The interval of two packets been sent for each thread */
private static final int INTERVAL = 80;
private static final String DATA = "5a76d93cb435fc54eba0b97156fe38f432a4e1da3a87cce8a222644466ed1317";
private class Sender implements Runnable {
private InetAddress address = null;
private DatagramSocket socket = null;
private String msg = null;
private String name = null;
private int packet_sent = 0;
public Sender(String addr, String msg, String name) throws SocketException,
UnknownHostException {
this.address = InetAddress.getByName(addr);
this.socket = new DatagramSocket();
this.msg = msg;
this.name = name;
}
@Override
public void run() {
// send request
byte[] buf = msg.getBytes();
DatagramPacket packet = new DatagramPacket(buf, buf.length,
address, 8000);
try {
for (int i = 0; i < PerformanceTest.PACKETS; i++) {
socket.send(packet);
packet_sent++;
Thread.sleep(INTERVAL);
}
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
//System.out.println("Thread " + name + " sends " + packet_sent + " packets.");
//System.out.println("Thread " + name + " end!");
}
}
/**
* @param args
*/
public static void main(String[] args) {
if (args.length != 1) {
System.out.println("Usage: java PerformanceTest <hostname>");
return;
}
String msg;
for (int i = 0; i < THREADS; i++) {
if(i % 2 == 0){
msg = i + "_" + (i+1) + ""r"n" + (i+1) + "_" + i + ""r"n" + DATA;
}else{
msg = i + "_" + (i-1) + ""r"n" + (i-1) + "_" + i + ""r"n" + DATA;
}
try {
new Thread(new PerformanceTest().new Sender(args[0], msg, "" + i)).start();
} catch (SocketException e) {
e.printStackTrace();
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
}
}
三、測(cè)試結(jié)果
測(cè)試環(huán)境：
Server：AMD Athlon(tm) 64 X2 Dual Core Processor 4000+，1G memory
Client：AMD Athlon(tm) 64 X2 Dual Core Processor 4000+，1G memory
在測(cè)試的過(guò)程中，當(dāng)INTERVAL設(shè)置的太小時(shí)，服務(wù)器端會(huì)出現(xiàn)丟包現(xiàn)象，INTERVAL越小，丟包越嚴(yán)重，為了提高Server的性能，特將Socket的ReceiveBufferSize設(shè)置成默認(rèn)大小的兩倍，
對(duì)于JDK實(shí)現(xiàn)：
public MediatorServerUseJDK(String name, int port) throws SocketException{
super(name);
socket = new DatagramSocket(port);
// set the receive buffer size to double default size
socket.setReceiveBufferSize(socket.getReceiveBufferSize() * 2);
System.out.println("Mediator server started on JDK model...");
System.out.println("Socket buffer size: " + socket.getReceiveBufferSize());
}
對(duì)于Mina實(shí)現(xiàn)：
DatagramAcceptorConfig acceptorConfig = acceptor.getDefaultConfig();
acceptorConfig.setThreadModel(ThreadModel.MANUAL);
// set the acceptor to reuse the address
acceptorConfig.getSessionConfig().setReuseAddress(true);
// set the receive buffer size to double default size
int recBufferSize = acceptorConfig.getSessionConfig().getReceiveBufferSize();
acceptorConfig.getSessionConfig().setReceiveBufferSize(recBufferSize * 2);
此時(shí)，相同的INTERVAL，丟包現(xiàn)象明顯減少。
接下來(lái)，再測(cè)試不同實(shí)現(xiàn)的性能差異：
UDP server started on JDK model...
Socket buffer size: 110592
INTERVAL ＝ 100ms，沒(méi)有出現(xiàn)丟包，
Process time: 49988
Process time: 49982
Process time: 49984
Process time: 49986
Process time: 49984
INTERVAL ＝ 80ms，仍然沒(méi)有丟包，不管Server是不是初次啟動(dòng)
Process time: 40006
Process time: 40004
Process time: 40003
Process time: 40005
Process time: 40013
UDP Server started on mina model...
Socket buffer size: 110592
INTERVAL ＝ 80ms，Server初次啟動(dòng)時(shí)，經(jīng)常會(huì)出現(xiàn)丟包，當(dāng)?shù)谝淮危ㄖ阜?wù)器初次啟動(dòng)時(shí)）沒(méi)有丟包時(shí)，隨后基本不丟包，
Process time: 39973
Process time: 40006
Process time: 40007
Process time: 40008
Process time: 40008
INTERVAL ＝ 100ms，沒(méi)有出現(xiàn)丟包
Process time: 49958
Process time: 49985
Process time: 49983
Process time: 49988
四、結(jié)論
在該要求下，采用JDK和Mina實(shí)現(xiàn)性能相當(dāng)，但是在Server初次啟動(dòng)時(shí)JDK實(shí)現(xiàn)基本不會(huì)出現(xiàn)丟包，而Mina實(shí)現(xiàn)則在Server初次啟動(dòng)時(shí)經(jīng)常出現(xiàn)丟包現(xiàn)象，在經(jīng)歷第一次測(cè)試后，兩種實(shí)現(xiàn)處理時(shí)間相近，請(qǐng)求并發(fā)量大概為每ms一個(gè)請(qǐng)求時(shí)，服務(wù)器不會(huì)出現(xiàn)丟包。