posts - 66, comments - 40, trackbacks - 0

本文將告訴你如何使用Netty2來編一個網(wǎng)絡(luò)應(yīng)用程序(包括客戶端和服務(wù)端)。我會介紹一個簡單的SumUp協(xié)議，用來對整數(shù)求和。通過源代碼的一步步講解，你會了解到Netty2的每個特性。

SumUp 協(xié)議

SumUp服務(wù)會加總從客戶端送來的ADD消息中的所有值，并且為每個ADD消息返回一個RESULT消息。所有消息都是由header和body兩部分組成：

header包含type和sequence兩個字段。type表示消息的類型(0是RESULT消息，1是ADD消息)。sequence用來表示一組對應(yīng)的ADD和RESULT(也就是說，服務(wù)器回應(yīng)ADD消息時，應(yīng)在RESULT中使用與ADD一樣的sequence值)。

ADD 消息

ADD消息包含了要被求和的值。

RESULT消息

RESULT具有不固定長度的消息體。當(dāng)計算沒問題時，body內(nèi)容是加總的值(4bytes)，如果有錯誤或溢位，則是2bytes. 見下圖：

實現(xiàn)MessageRecognizer

MessageRecognizer從送來的數(shù)據(jù)中重組出Message對象。這兒我們實現(xiàn)了一個SumUpMessageRecognizer，用于客戶端和服務(wù)端的信息重組。

public class SumUpMessageRecognizer implements MessageRecognizer {
??
?? publicstaticfinalint CLIENT_MODE = 1;
??
?? publicstaticfinalint SERVER_MODE = 2;
??
?? privateint mode;
??
?? publicSumUpMessageRecognizer(int mode) {
??? ??? switch (mode) {
??? ??????? case CLIENT_MODE:
??????? ??? ??????? case SERVER_MODE:
???? ??????????? this.mode = mode;
???? ??????????? break;
???????????????????? default:
???? ??????????? thrownew IllegalArgumentException("invalid mode: " + mode);
??? ??? }
?? }
??
?? public Message recognize(ByteBuffer buf) throws MessageParseException {
??? ??? // return null if message type is not arrived yet.
??? ??? if (buf.remaining() < Constants.TYPE_LEN)
???? ??????? returnnull;
????
??? ??? int type = buf.getShort();
??? ??? switch (mode) {
??? ?????? ?// 如果是server模式，只讓它接收ADD消息.
??? ??????? case SERVER_MODE:
???? ?????????? ?switch (type) {
???? ??????????????? case Constants.ADD:
????? ??????????????? returnnewAddMessage();
???? ??????????????? default:
????? ??????????????????? thrownew MessageParseException("unknown type: " + type);
???? ??????????? }
??? ??????? //如果是客戶端模式，只讓它接收RESULT消息.
??? ??????? case CLIENT_MODE:
???????????????????????? switch (type) {
???? ??????????????? case Constants.RESULT:
????? ??????????????????? returnnewResultMessage();
???? ??????????????? default:
????? ??????????????????????? thrownew MessageParseException("unknown type: " + type);
???? ??????????? }
??? ??????? default:
???? ??????????? thrownew InternalError(); // this cannot happen
??? ??? }
?? }
? }

實現(xiàn)ADD和RESULT消息

我們必須實現(xiàn)ADD和RESULT消息: ADD和RESULT。它們都有公共的header，最好的方式是實現(xiàn)一個AbstractMessage，并且從它繼承出Add和Result消息。

源代碼：

AbstractMessage.java

View Javadoc

						1
						/*

						2
						 * @(#) $Id: AbstractMessage.java 11 2005-04-18 03:42:45Z trustin $
						 */
						package 
						net.gleamynode.netty2.example.sumup;
56import java.nio.ByteBuffer;
78import net.gleamynode.netty2.Message;
9import net.gleamynode.netty2.MessageParseException;

						17
						public
						abstract
						class
						AbstractMessage implements Message {
1819privatefinalint type;
2021privateint sequence;
2223privateboolean readHeader;
2425privateboolean wroteHeader;
2627protectedAbstractMessage(int type) {
28this.type = type;
29  	}
3031publicint getSequence() {
32return sequence;
33  	}
3435publicvoid setSequence(int sequence) {
36this.sequence = sequence;
37  	}
3839publicfinalboolean read(ByteBuffer buf) throws MessageParseException {
40// read a header if not read yet.41if (!readHeader) {
42  			readHeader = readHeader(buf);
43if (!readHeader)
44return false;
45  		}
4647// Header is read, now try to read body48if (readBody(buf)) {
49// finished reading single complete message50

		readHeader = false; // reset state51returntrue;
52  		} else53return false;
54  	}
5556privateboolean readHeader(ByteBuffer buf) throws MessageParseException {
57// if header is not fully read, don't read it.58if (buf.remaining() < Constants.HEADER_LEN)
59return false;
6061// read header and validate the message62int readType = buf.getShort();
63if (type != readType)
64thrownew MessageParseException("type mismatches: " + readType
65  					+ " (expected: " + type + ')');
6667// read sequence number of the message68  		sequence = buf.getInt();
69returntrue;
70  	}
7172protectedabstractboolean readBody(ByteBuffer buf)
73  			throws MessageParseException;
7475publicboolean write(ByteBuffer buf) {
76// write a header if not written yet.77if (!wroteHeader) {
78  			wroteHeader = writeHeader(buf);
79if (!wroteHeader)
80return false; // buffer is almost full perhaps81  		}
8283// Header is written, now try to write body84if (writeBody(buf)) {
85// finished writing single complete message86  			wroteHeader = false;
87returntrue;
88  		} else {
89return false;
90  		}
91  	}
9293privateboolean writeHeader(ByteBuffer buf) {
94// check if there is enough space to write header95if (buf.remaining() < Constants.HEADER_LEN) return false;
96  		buf.putShort((short) type);
97  		buf.putInt(sequence);
98returntrue;
99  	}
100101protectedabstractboolean writeBody(ByteBuffer buf);
102 }

AddMessage.java

View Javadoc

						
						
						package 
						
								net.gleamynode.netty2.example.sumup
						
						
								5
						
						
								6
						
						import java.nio.ByteBuffer;
78import net.gleamynode.netty2.MessageParseException;
publicclassAddMessageextendsAbstractMessage {
1718privateint value;
1920publicAddMessage() {
21super(Constants.ADD);
22  	}
2324publicint getValue() {
25return value;
26  	}
2728publicvoid setValue(int value) {
29this.value = value;
30  	}
3132protectedboolean readBody(ByteBuffer buf) throws MessageParseException {
33// don't read body if it is partially readable34if (buf.remaining() < Constants.ADD_BODY_LEN) return false;
35  		value = buf.getInt();
36returntrue;
37  	}
3839protectedboolean writeBody(ByteBuffer buf) {
40// check if there is enough space to write body41if (buf.remaining() < Constants.ADD_BODY_LEN)
42return false;
4344  		buf.putInt(value);
4546returntrue;
47  	}
4849public String toString() {
50// it is a good practice to create toString() method on message classes.51return getSequence() + ":ADD(" + value + ')';
52  	}
53  }

ResultMessage.jav a?

View Javadoc

package net.gleamynode.netty2.example.sumup 5 6 import java.nio.ByteBuffer; 7 8import net.gleamynode.netty2.MessageParseException;

16 public class ResultMessage extends AbstractMessage { 17 18privateboolean ok; 19 20privateint value; 21 22privateboolean processedResultCode; 23 24publicResultMessage() { 25super(Constants.RESULT); 26 } 27 28publicboolean isOk() { 29return ok; 30 } 31 32publicvoid setOk(boolean ok) { 33this.ok = ok; 34 } 35 36publicint getValue() { 37return value; 38 } 39 40publicvoid setValue(int value) { 41this.value = value; 42 } 43 44protectedboolean readBody(ByteBuffer buf) throws MessageParseException { 45if (!processedResultCode) { 46 processedResultCode = readResultCode(buf); 47if (!processedResultCode) 48return false; 49 } 50 51if (ok) { 52if (readValue(buf)) { 53 processedResultCode = false; 54returntrue; 55 } else56return false; 57 } else { 58 processedResultCode = false; 59returntrue; 60 } 61 } 62 63privateboolean readResultCode(ByteBuffer buf) { 64if (buf.remaining() < Constants.RESULT_CODE_LEN) 65return false; 66 ok = buf.getShort() == Constants.RESULT_OK; 67returntrue; 68 } 69 70privateboolean readValue(ByteBuffer buf) { 71if (buf.remaining() < Constants.RESULT_VALUE_LEN) 72return false; 73 value = buf.getInt(); 74returntrue; 75 76 } 77 78protectedboolean writeBody(ByteBuffer buf) { 79// check if there is enough space to write body80if (buf.remaining() < Constants.RESULT_CODE_LEN 81 + Constants.RESULT_VALUE_LEN) 82return false; 83 84 buf.putShort((short) (ok ? Constants.RESULT_OK 86 : Constants.RESULT_ERROR)); 87if (ok) 88 buf.putInt(value); 89 90returntrue; 91 } 92 93public String toString() { 94if (ok) { 95return getSequence() + ":RESULT(" + value + ')'; 96 } else { 97return getSequence() + ":RESULT(ERROR)"; 98 } 99 } 100 }

This page was automatically generated by Maven

實現(xiàn)協(xié)議處理流程

實現(xiàn)了Messagerecognizer和Message之后，要實現(xiàn)Server和Client是非常容易的事情，通過下面的代碼，你會很容易理解如何去實現(xiàn)協(xié)議的處理流程。

實現(xiàn)Server

實現(xiàn)服務(wù)端兩個主要的類，一個是Server類，另一個是ServerSessionListener. Server類負(fù)責(zé)啟動主程序并監(jiān)聽連接。而ServerSessionListener用于處理和發(fā)送消息。

public class Server {

privatestaticfinalint SERVER_PORT = 8080;
privatestaticfinalint DISPATCHER_THREAD_POOL_SIZE = 16;

publicstaticvoid main(String[] args) throws Throwable {
//?初始化 I/O processor?和 event dispatcher
IoProcessor ioProcessor = new IoProcessor();
ThreadPooledEventDispatcher eventDispatcher = new OrderedEventDispatcher();
?
? // 啟動缺省數(shù)量的I/O工作線程
? ioProcessor.start();
?
? // 啟動指定數(shù)量的event dispatcher?線程
? eventDispatcher.setThreadPoolSize(DISPATCHER_THREAD_POOL_SIZE);
? eventDispatcher.start();
?
? //?準(zhǔn)備 message recognizer
? MessageRecognizer recognizer = new SumUpMessageRecognizer(
? SumUpMessageRecognizer.SERVER_MODE);
?
? // 準(zhǔn)備session監(jiān)聽器，用于處理通訊過程.
?ServerSessionListener listener = new ServerSessionListener();
?
?// 開啟server socket通道
?ServerSocketChannel ssc = ServerSocketChannel.open();
?ssc.socket().bind(new InetSocketAddress(SERVER_PORT));
?
?// 監(jiān)聽連接，并開始通訊
?System.out.println("listening on port " + SERVER_PORT);
?for (;;) {
?// 接受connection
?SocketChannel channel = ssc.accept();
?
?//?建立新的session
?Session session = new Session(ioProcessor, channel, recognizer, eventDispatcher);
?
?// 添加session監(jiān)聽器
?session.addSessionListener(listener);
?
?//?開始通訊
?session.start();
?}
?}
?}
?

public class ServerSessionListener implements SessionListener {

??????? public ServerSessionListener() {

??????? }

??????? public void connectionEstablished(Session session) {

?????????????? System.out.println(session.getSocketAddress() + " connected");

?????????????? //?設(shè)置空閑時間為60秒

?????????????? session.getConfig().setIdleTime(60);

?????????????? // 設(shè)置sum的初始值為0。

?????????????? session.setAttachment(new Integer(0));

??????? }

??????? public void connectionClosed(Session session) {

?????????????? System.out.println(session.getSocketAddress() + " closed");

? ?????? }

??????? // 當(dāng)收到client發(fā)來的消息時，此方法被調(diào)用

? ?????? public void messageReceived(Session session, Message message) {

? ????????????? System.out.println(session.getSocketAddress() + " RCVD: " + message);

?????????????? // client端只發(fā)送AddMessage. 其它情況要另作處理

?????????????? // 在這里只是簡單的進(jìn)行類型轉(zhuǎn)換處理

?????????????? AddMessage am = (AddMessage) message;

?????????????? // 將收到的消息里的值加上當(dāng)前sum的值.

?????????????? int sum = ((Integer) session.getAttachment()).intValue();

?????????????? int value = am.getValue();

?????????????? long expectedSum = (long) sum + value;

?????????????? if (expectedSum > Integer.MAX_VALUE || expectedSum < Integer.MIN_VALUE) {

?????????????????????? // 如果溢位返回錯誤消息

??????? ?????????????? ResultMessage rm = newResultMessage();

??????? ?????????????? rm.setSequence(am.getSequence()); // 從送來的Add消息中得到sequence值。

?????????????????????? rm.setOk(false);

?????????????????????? session.write(rm);

?????????????? } else {

?????????????????????? //? 加總

?????????????????????? sum = (int) expectedSum;

?????????????????????? session.setAttachment(new Integer(sum));

?????????????????????? // 返回結(jié)果消息

?????????????????????? ResultMessage rm = newResultMessage();

?????????????????????? rm.setSequence(am.getSequence()); // 從送來的Add消息中得到sequence值。

?????????????????????? rm.setOk(true);

?????????????????????? rm.setValue(sum);

?????????????????????? session.write(rm);

?????????????? }

??????? }

??????? public void messageSent(Session session, Message message) {

?????????????? System.out.println(session.getSocketAddress() + " SENT: " + message);

??????? }

??????? public void sessionIdle(Session session) {

?????????????? System.out.println(session.getSocketAddress()

?????????????????????????????? + " disconnecting the idle");

?????????????? // 關(guān)閉空閑的會話。

?????????????? session.close();

??????? }

??????? // 異常發(fā)生時，將調(diào)用此方法

??????? public void exceptionCaught(Session session, Throwable cause) {

?????????????? System.out.println(Thread.currentThread().getName()

?????????????????????????????? + session.getSocketAddress() + " exception:");

? ????????????? cause.printStackTrace(System.out);

? ????????????? if (cause instanceof MessageParseException) {

? ????????????????????? //?印出錯誤信息內(nèi)容，便于調(diào)試

?????????????????????? MessageParseException mpe = (MessageParseException) cause;

?????????????????????? ByteBuffer buf = mpe.getBuffer();

?????????????????????? System.out.println(buf);

?????????????????????? System.out.print("Buffer Content: ");

?????????????????????? while (buf.remaining() > 0) {

?????????????????????????????? System.out.print(buf.get() & 0xFF);

?????????????????????????????? System.out.print(' ');

?????????????????????? }

?????????????????????? System.out.println();

?????????????? }

?????????????? // 關(guān)閉會話

?????????????? session.close();

??????? }

}

服務(wù)端運(yùn)行后，其輸出的內(nèi)容示例如下：

listening on port 8080
/127.0.0.1:4753 connected
/127.0.0.1:4753 RCVD: 0:ADD(4)
/127.0.0.1:4753 RCVD: 1:ADD(6)
/127.0.0.1:4753 RCVD: 2:ADD(2)
/127.0.0.1:4753 RCVD: 3:ADD(7)
/127.0.0.1:4753 RCVD: 4:ADD(8)
/127.0.0.1:4753 RCVD: 5:ADD(1)
/127.0.0.1:4753 SENT: 0:RESULT(4)
/127.0.0.1:4753 SENT: 1:RESULT(10)
/127.0.0.1:4753 SENT: 2:RESULT(12)
/127.0.0.1:4753 SENT: 3:RESULT(19)
/127.0.0.1:4753 SENT: 4:RESULT(27)
/127.0.0.1:4753 SENT: 5:RESULT(28)
/127.0.0.1:4753 closed

														實現(xiàn)客戶端

跟服務(wù)端對應(yīng)，主要由Client和ClientSessionListener組成。

public class Client {

??????? private static final String HOSTNAME = "localhost";

??????? private static final int PORT = 8080;

??????? private static final int CONNECT_TIMEOUT = 30; // seconds

??????? private static final int DISPATCHER_THREAD_POOL_SIZE = 4;

??????? public static void main(String[] args) throws Throwable {

?????????????? // 預(yù)備要加總的值。

?????????????? int[] values = newint[args.length];

?????????????? for (int i = 0; i < args.length; i++) {

?????????????????????? values[i] = Integer.parseInt(args[i]);

?????????????? }

?????????????? // 初始化 I/O processor 和 event dispatcher

?????????????? IoProcessor ioProcessor = new IoProcessor();

?????????????? ThreadPooledEventDispatcher eventDispatcher = new OrderedEventDispatcher();

?????????????? // 開始缺省數(shù)量的I/O工作線程

?????????????? ioProcessor.start();

?????????????? // 啟動指定數(shù)量的event dispatcher線程

??????? eventDispatcher.setThreadPoolSize(DISPATCHER_THREAD_POOL_SIZE

?????????????? eventDispatcher.start();

?????????????? // 準(zhǔn)備 message recognizer

?????????????? MessageRecognizer recognizer = new SumUpMessageRecognizer(

?????????????????????????????? SumUpMessageRecognizer.CLIENT_MODE);

?????????????? // 準(zhǔn)備客戶端會話。

?????????????? Session session = new Session(ioProcessor, new InetSocketAddress(

?????????????????????????????? HOSTNAME, PORT), recognizer, eventDispatcher);

??????????????

?????????????? session.getConfig().setConnectTimeout(CONNECT_TIMEOUT);

??????????????

?????????????? // 開始會話，并使用ClientSessionListener監(jiān)聽。

?????????????? ClientSessionListener listener = new ClientSessionListener(values);

?????????????? session.addSessionListener(listener);

?????????????? session.start();

??????????????

?????????????? // 一直等到加總完成

?????????????? while ( !listener.isComplete() ) {

?????????????????????? Thread.sleep(1000);

?????????????? }

??????????????

?????????????? // 停止 I/O processor 和 event dispatcher

?????????????? eventDispatcher.stop();

?????????????? ioProcessor.stop();

? ?????? }

public class ClientSessionListener implements SessionListener {

? ?????? private final int[] values;

? ?????? private boolean complete;

? ?????? public ClientSessionListener(int[] values) {

? ????????????? this.values = values;

? ?????? }

? ??????

? ?????? public boolean isComplete() {

? ????????????? return complete;

? ?????? }

??????? // 當(dāng)連接建立好后會調(diào)用此方法。

? ?????? public void connectionEstablished(Session session) {

? ????????????? System.out.println("connected to " + session.getSocketAddress());

? ????????????? // 發(fā)送加總請求。

? ????????????? for (int i = 0; i < values.length; i++) {

? ????????????????????? AddMessage m = new AddMessage();

? ????????????????????? m.setSequence(i);

? ????????????????????? m.setValue(values[i]);

? ????????????????????? session.write(m);

? ????????????? }

? ?????? }

? ?????? public void connectionClosed(Session session) {

? ????????????? System.out.println("disconnected from " + session.getSocketAddress());

??????? }

??????? // 當(dāng)收到server的回應(yīng)信息時，會調(diào)用此方法

??????? public void messageReceived(Session session, Message message) {

?????????????? System.out.println("RCVD: " + message);

?????????????? // 服務(wù)端只發(fā)送ResultMessage. 其它情況下

? ????????????? // 要通過instanceOf來判斷它的類型.

? ????????????? ResultMessage rm = (ResultMessage) message;

? ????????????? if (rm.isOk()) {

? ????????????????????? // 如果ResultMessage是OK的.

? ????????????????????? // 根據(jù)ResultMessage的sequence值來判斷如果,

? ????????????????????? // 一次消息的sequence值，則

? ????????????????????? if (rm.getSequence() == values.length - 1) {

? ????????????????????????????? // 打印出結(jié)果.

? ????????????????????????????? System.out.println("The sum: " + rm.getValue());
???????????????????????????????// 關(guān)閉會話

?????????????????????????????? session.close();

?????????????????????????????? complete = true;

?????????????????????? }

?????????????? } else {

?????????????????????? //?如有錯誤，則打印錯誤信息，并結(jié)束會話.

? ????????????????????? System.out.println("server error, disconnecting...");

?????????????????????? session.close();

?????????????????????? complete = true;

?????????????? }

??????? }

??????? public void messageSent(Session session, Message message) {

?????????????? System.out.println("SENT: " + message);

??????? }

??????? public void sessionIdle(Session session) {

??????????????

??????? }

??????? public void exceptionCaught(Session session, Throwable cause) {

?????????????? cause.printStackTrace(System.out);

?????????????? if (cause instanceof ConnectException) {

?????????????????????? // 如果連接server失敗, 則間隔5秒重試連接.

? ????????????????????? System.out.println("sleeping...");

? ????????????????????? try {

? ????????????????????????????? Thread.sleep(5000);

? ????????????????????? } catch (InterruptedException e) {

? ????????????????????? }

? ?????????????????????

? ????????????????????? System.out.println("reconnecting... " + session.getSocketAddress());

? ????????????????????? session.start();

? ????????????? } else {

? ????????????????????? session.close();

? ????????????? }

? ?????? }

通過上面的例子，你也許會發(fā)現(xiàn)實現(xiàn)一個自定義的協(xié)議原來如此簡單。你如果用Netty試著去實現(xiàn)自己的smtp或pop協(xié)議，我想也不會是一件難事了。

Netty2的首頁在http://gleamynode.net/dev/projects/netty2/index.html，你可以在這找到本文的全部源碼。

posted on 2006-06-20 19:45 happytian 閱讀(370) 評論(0) 編輯收藏

新用戶注冊刷新評論列表


只有注冊用戶登錄后才能發(fā)表評論。




網(wǎng)站導(dǎo)航: 博客園 IT新聞 Chat2DB C++博客博問管理

2025年5月

日

一

二

三

四

五

六

Welcome here, my friend!

常用鏈接

留言簿(12)

隨筆檔案(66)

文章分類

人生

文章檔案(63)

web

Ajax forum
java.csdn
javafan
Mrs Hu's blog
sohu

常用鏈接

留言簿(12)

隨筆檔案(66)

文章分類

文章檔案(63)

web

最新隨筆

搜索

積分與排名

最新評論

閱讀排行榜

評論排行榜