posts - 66, comments - 40, trackbacks - 0

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

SumUp 協(xié)議

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

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

ADD 消息

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

RESULT消息

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

實(shí)現(xiàn)MessageRecognizer

MessageRecognizer從送來(lái)的數(shù)據(jù)中重組出Message對(duì)象。這兒我們實(shí)現(xiàn)了一個(gè)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
??? ??? }
?? }
? }

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

我們必須實(shí)現(xiàn)ADD和RESULT消息: ADD和RESULT。它們都有公共的header，最好的方式是實(shí)現(xiàn)一個(gè)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

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

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

實(shí)現(xiàn)Server

實(shí)現(xiàn)服務(wù)端兩個(gè)主要的類，一個(gè)是Server類，另一個(gè)是ServerSessionListener. Server類負(fù)責(zé)啟動(dòng)主程序并監(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();
?
? // 啟動(dòng)缺省數(shù)量的I/O工作線程
? ioProcessor.start();
?
? // 啟動(dòng)指定數(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è)置空閑時(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ā)來(lái)的消息時(shí)，此方法被調(diào)用

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

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

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

?????????????? // 在這里只是簡(jiǎn)單的進(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) {

?????????????????????? // 如果溢位返回錯(cuò)誤消息

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

??????? ?????????????? rm.setSequence(am.getSequence()); // 從送來(lái)的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()); // 從送來(lái)的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)閉空閑的會(huì)話。

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

??????? }

??????? // 異常發(fā)生時(shí)，將調(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) {

? ????????????????????? //?印出錯(cuò)誤信息內(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)閉會(huì)話

?????????????? 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

														實(shí)現(xiàn)客戶端

跟服務(wù)端對(duì)應(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();

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

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

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

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

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

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

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

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

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

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

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

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

?????????????? // 開始會(huì)話，并使用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)連接建立好后會(huì)調(diào)用此方法。

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

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

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

? ????????????? 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)信息時(shí)，會(huì)調(diào)用此方法

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

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

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

? ????????????? // 要通過instanceOf來(lái)判斷它的類型.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

? ?????? }

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

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

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

新用戶注冊(cè) 刷新評(píng)論列表


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




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

2025年7月

日

一

二

三

四

五

六

Welcome here, my friend!

常用鏈接

留言簿(12)

隨筆檔案(66)

文章分類

人生

文章檔案(63)

web

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

常用鏈接

留言簿(12)

隨筆檔案(66)

文章分類

文章檔案(63)

web

最新隨筆

搜索

積分與排名

最新評(píng)論

閱讀排行榜

評(píng)論排行榜