Apache Mina使用手记（四）

港控/mmm° 2022-08-13 12:48 181阅读 0赞

上一篇中，我们介绍了如何在mina中编写自己的日志过滤器，这一篇我们自己实现一个编解器。

实际应用当，很多应用系统应用的都不是标准的web service或XML等，比如象中国移动/联通/电信的短信网关程序，都有自己不同的协议实现，并且都是基于TCP/IP的字节流。Mina自带的编解码器实现了TextLineEncoder和TextLineDecoder，可以进行按行的字符串处理，对于象短信网关程序，就要自己实现编解码过滤器了。

我们定义一个简单的基于TCP/IP字节流的协议，实现在客户端和服务端之间的数据包传输。数据包MyProtocalPack有消息头和消息体组成，消息头包括：length（消息包的总长度，数据类型int），flag（消息包标志位，数据类型byte），消息体content是一个字符串，实际实现的时候按byte流处理。源代码如下：

**\[java\]**  [view plain][] [copy][view plain]

1.  **package** com.gftech.mytool.mina;  
2.  **import** com.gftech.util.GFCommon;  
3.  **public** **class** MyProtocalPack \{  
4.      **private** **int** length;  
5.      **private** **byte** flag;  
6.      **private** String content;  
7.        
8.      **public** MyProtocalPack()\{  
9.            
10.     \}  
11.       
12.     **public** MyProtocalPack(**byte** flag,String content)\{  
13.         **this**.flag=flag;  
14.         **this**.content=content;  
15.         **int** len1=content==**null**?0:content.getBytes().length;  
16.         **this**.length=5\+len1;  
17.     \}  
18.       
19.     **public** MyProtocalPack(**byte**\[\] bs)\{  
20.         **if**(bs!=**null** && bs.length>=5)\{  
21.             length=GFCommon.bytes2int(GFCommon.bytesCopy(bs, 0, 4));  
22.             flag=bs\[4\];  
23.             content=**new** String(GFCommon.bytesCopy(bs, 5, length-5));  
24.         \}  
25.     \}  
26.       
27.     **public** **int** getLength() \{  
28.         **return** length;  
29.     \}  
30.     **public** **void** setLength(**int** length) \{  
31.         **this**.length = length;  
32.     \}  
33.     **public** **byte** getFlag() \{  
34.         **return** flag;  
35.     \}  
36.     **public** **void** setFlag(**byte** flag) \{  
37.         **this**.flag = flag;  
38.     \}  
39.     **public** String getContent() \{  
40.         **return** content;  
41.     \}  
42.     **public** **void** setContent(String content) \{  
43.         **this**.content = content;  
44.     \}  
45.       
46.     **public** String toString()\{  
47.         StringBuffer sb=**new** StringBuffer();  
48.         sb.append(" Len:").append(length);  
49.         sb.append(" flag:").append(flag);  
50.         sb.append(" content:").append(content);  
51.         **return** sb.toString();  
52.     \}  
53. \}

回过头来，我们先看一下在MinaTimeServer中，如何使用一个文本的编解码过滤器，它是在过滤器链中添加了一个叫ProtocalCodecFilter的类，其中它调用 了一个工厂方法TextLineCodecFactory的工厂类，创建具休的TextLineEncoder和TextLineDecoder编码和解 码器。我们看一下具体的源代码：

**\[java\]**  [view plain][] [copy][view plain]

1.  acceptor.getFilterChain().addLast("codec", **new** ProtocolCodecFilter(**new** TextLineCodecFactory(Charset.forName("GBK"))));

**\[java\]**  [view plain][] [copy][view plain]

1. **package** org.apache.mina.filter.codec.textline;  
2. **import** java.nio.charset.Charset;  
3. **import** org.apache.mina.core.buffer.BufferDataException;  
4. **import** org.apache.mina.core.session.IoSession;  
5. **import** org.apache.mina.filter.codec.ProtocolCodecFactory;  
6. **import** org.apache.mina.filter.codec.ProtocolDecoder;  
7. **import** org.apache.mina.filter.codec.ProtocolEncoder;  
8. /\*\* 
9.  \* A \{@link ProtocolCodecFactory\} that performs encoding and decoding between 
10.  \* a text line data and a Java string object.  This codec is useful especially 
11.  \* when you work with a text-based protocols such as SMTP and IMAP. 
12.  \* 
13.  \* @author The Apache MINA Project (dev@mina.apache.org) 
14.  \* @version $Rev$, $Date$ 
15.  \*/  
16. **public** **class** TextLineCodecFactory **implements** ProtocolCodecFactory \{  
17.     **private** **final** TextLineEncoder encoder;  
18.     **private** **final** TextLineDecoder decoder;  
19.     /\*\* 
20.      \* Creates a new instance with the current default \{@link Charset\}. 
21.      \*/  
22.     **public** TextLineCodecFactory() \{  
23.         **this**(Charset.defaultCharset());  
24.     \}  
25.     /\*\* 
26.      \* Creates a new instance with the specified \{@link Charset\}.  The 
27.      \* encoder uses a UNIX \{@link LineDelimiter\} and the decoder uses 
28.      \* the AUTO \{@link LineDelimiter\}. 
29.      \* 
30.      \* @param charset 
31.      \*  The charset to use in the encoding and decoding 
32.      \*/  
33.     **public** TextLineCodecFactory(Charset charset) \{  
34.         encoder = **new** TextLineEncoder(charset, LineDelimiter.UNIX);  
35.         decoder = **new** TextLineDecoder(charset, LineDelimiter.AUTO);  
36.     \}  
37.     /\*\* 
38.      \* Creates a new instance of TextLineCodecFactory.  This constructor 
39.      \* provides more flexibility for the developer. 
40.      \* 
41.      \* @param charset 
42.      \*  The charset to use in the encoding and decoding 
43.      \* @param encodingDelimiter 
44.      \*  The line delimeter for the encoder 
45.      \* @param decodingDelimiter 
46.      \*  The line delimeter for the decoder 
47.      \*/  
48.     **public** TextLineCodecFactory(Charset charset,  
49.             String encodingDelimiter, String decodingDelimiter) \{  
50.         encoder = **new** TextLineEncoder(charset, encodingDelimiter);  
51.         decoder = **new** TextLineDecoder(charset, decodingDelimiter);  
52.     \}  
53.     /\*\* 
54.      \* Creates a new instance of TextLineCodecFactory.  This constructor 
55.      \* provides more flexibility for the developer. 
56.      \* 
57.      \* @param charset 
58.      \*  The charset to use in the encoding and decoding 
59.      \* @param encodingDelimiter 
60.      \*  The line delimeter for the encoder 
61.      \* @param decodingDelimiter 
62.      \*  The line delimeter for the decoder 
63.      \*/  
64.     **public** TextLineCodecFactory(Charset charset,  
65.             LineDelimiter encodingDelimiter, LineDelimiter decodingDelimiter) \{  
66.         encoder = **new** TextLineEncoder(charset, encodingDelimiter);  
67.         decoder = **new** TextLineDecoder(charset, decodingDelimiter);  
68.     \}  
69.     **public** ProtocolEncoder getEncoder(IoSession session) \{  
70.         **return** encoder;  
71.     \}  
72.     **public** ProtocolDecoder getDecoder(IoSession session) \{  
73.         **return** decoder;  
74.     \}  
75.        /\*\* 
76.      \* Returns the allowed maximum size of the encoded line. 
77.      \* If the size of the encoded line exceeds this value, the encoder 
78.      \* will throw a \{@link IllegalArgumentException\}.  The default value 
79.      \* is \{@link Integer\#MAX\_VALUE\}. 
80.      \*  
81.      \* This method does the same job with \{@link TextLineEncoder\#getMaxLineLength()\}. 
82.      \*/  
83.     **public** **int** getEncoderMaxLineLength() \{  
84.         **return** encoder.getMaxLineLength();  
85.     \}  
86.     /\*\* 
87.      \* Sets the allowed maximum size of the encoded line. 
88.      \* If the size of the encoded line exceeds this value, the encoder 
89.      \* will throw a \{@link IllegalArgumentException\}.  The default value 
90.      \* is \{@link Integer\#MAX\_VALUE\}. 
91.      \*  
92.      \* This method does the same job with \{@link TextLineEncoder\#setMaxLineLength(int)\}. 
93.      \*/  
94.     **public** **void** setEncoderMaxLineLength(**int** maxLineLength) \{  
95.         encoder.setMaxLineLength(maxLineLength);  
96.     \}  
97.     /\*\* 
98.      \* Returns the allowed maximum size of the line to be decoded. 
99.      \* If the size of the line to be decoded exceeds this value, the 
100.      \* decoder will throw a \{@link BufferDataException\}.  The default 
101.      \* value is <tt>1024</tt> (1KB). 
102.      \*  
103.      \* This method does the same job with \{@link TextLineDecoder\#getMaxLineLength()\}. 
104.      \*/  
105.     **public** **int** getDecoderMaxLineLength() \{  
106.         **return** decoder.getMaxLineLength();  
107.     \}  
108.     /\*\* 
109.      \* Sets the allowed maximum size of the line to be decoded. 
110.      \* If the size of the line to be decoded exceeds this value, the 
111.      \* decoder will throw a \{@link BufferDataException\}.  The default 
112.      \* value is <tt>1024</tt> (1KB). 
113.      \*  
114.      \* This method does the same job with \{@link TextLineDecoder\#setMaxLineLength(int)\}. 
115.      \*/  
116.     **public** **void** setDecoderMaxLineLength(**int** maxLineLength) \{  
117.         decoder.setMaxLineLength(maxLineLength);  
118.     \}  
119. \}

TextLineFactory实现了ProtocalCodecFactory接口，该接口主要有一个编码的方法getEncoder()和一个解码的方法getDecoder()：

**\[java\]**  [view plain][] [copy][view plain]

1. **package** org.apache.mina.filter.codec;  
2. **import** org.apache.mina.core.session.IoSession;  
3. /\*\* 
4.  \* Provides \{@link ProtocolEncoder\} and \{@link ProtocolDecoder\} which translates 
5.  \* binary or protocol specific data into message object and vice versa. 
6.  \*  
7.  \* Please refer to 
8.  \* <a href="../../../../../xref-examples/org/apache/mina/examples/reverser/ReverseProtocolProvider.html" mce\_href="xref-examples/org/apache/mina/examples/reverser/ReverseProtocolProvider.html"><code>ReverserProtocolProvider</code></a> 
9.  \* example. 
10.  \* 
11.  \* @author The Apache MINA Project (dev@mina.apache.org) 
12.  \* @version $Rev$, $Date$ 
13.  \*/  
14. **public** **interface** ProtocolCodecFactory \{  
15.     /\*\* 
16.      \* Returns a new (or reusable) instance of \{@link ProtocolEncoder\} which 
17.      \* encodes message objects into binary or protocol-specific data. 
18.      \*/  
19.     ProtocolEncoder getEncoder(IoSession session) **throws** Exception;  
20.     /\*\* 
21.      \* Returns a new (or reusable) instance of \{@link ProtocolDecoder\} which 
22.      \* decodes binary or protocol-specific data into message objects. 
23.      \*/  
24.     ProtocolDecoder getDecoder(IoSession session) **throws** Exception;  
25. \}

我们主要是仿照TextLineEncoder实现其中的encode（）方法，仿照TextLineDecoder实现其中的decode（）即可，它们分别实现了ProtocalEncoder和ProtocalDecoder接口。我们要编写三个类分别是：MyProtocalCodecFactory，MyProtocalEncoder，MyProtocalDecoder对应TextLineCodecFactory，TextLineEncoder，TextLineDecoder。

**\[java\]**  [view plain][] [copy][view plain]

1.  **package** com.gftech.mytool.mina;  
2.  **import** java.nio.charset.Charset;  
3.  **import** org.apache.mina.core.session.IoSession;  
4.  **import** org.apache.mina.filter.codec.ProtocolCodecFactory;  
5.  **import** org.apache.mina.filter.codec.ProtocolDecoder;  
6.  **import** org.apache.mina.filter.codec.ProtocolEncoder;  
7.  **public** **class** MyProtocalCodecFactory   **implements** ProtocolCodecFactory \{  
8.          **private** **final** MyProtocalEncoder encoder;  
9.          **private** **final** MyProtocalDecoder decoder;  
10.           
11.         **public** MyProtocalCodecFactory(Charset charset) \{  
12.             encoder=**new** MyProtocalEncoder(charset);  
13.             decoder=**new** MyProtocalDecoder(charset);  
14.         \}  
15.            
16.         **public** ProtocolEncoder getEncoder(IoSession session) \{  
17.             **return** encoder;  
18.         \}  
19.         **public** ProtocolDecoder getDecoder(IoSession session) \{  
20.             **return** decoder;  
21.         \}  
22.           
23. \}

**\[java\]**  [view plain][] [copy][view plain]

1.  **package** com.gftech.mytool.mina;  
2.  **import** java.nio.charset.Charset;  
3.  **import** org.apache.mina.core.buffer.IoBuffer;  
4.  **import** org.apache.mina.core.session.IoSession;  
5.  **import** org.apache.mina.filter.codec.ProtocolEncoderAdapter;  
6.  **import** org.apache.mina.filter.codec.ProtocolEncoderOutput;  
7.  **public** **class** MyProtocalEncoder **extends** ProtocolEncoderAdapter \{  
8.      **private** **final** Charset charset;  
9.      **public** MyProtocalEncoder(Charset charset) \{  
10.         **this**.charset = charset;  
11.     \}  
12.     //在此处实现对MyProtocalPack包的编码工作，并把它写入输出流中  
13.     **public** **void** encode(IoSession session, Object message, ProtocolEncoderOutput out) **throws** Exception \{  
14.         MyProtocalPack value = (MyProtocalPack) message;  
15.         IoBuffer buf = IoBuffer.allocate(value.getLength());  
16.         buf.setAutoExpand(**true**);  
17.         buf.putInt(value.getLength());  
18.         buf.put(value.getFlag());  
19.         **if** (value.getContent() != **null**)  
20.             buf.put(value.getContent().getBytes());  
21.         buf.flip();  
22.         out.write(buf);  
23.     \}  
24.     **public** **void** dispose() **throws** Exception \{  
25.     \}  
26. \}

**\[java\]**  [view plain][] [copy][view plain]

1. **package** com.gftech.mytool.mina;  
2. **import** java.nio.charset.Charset;  
3. **import** java.nio.charset.CharsetDecoder;  
4. **import** org.apache.mina.core.buffer.IoBuffer;  
5. **import** org.apache.mina.core.session.AttributeKey;  
6. **import** org.apache.mina.core.session.IoSession;  
7. **import** org.apache.mina.filter.codec.ProtocolDecoder;  
8. **import** org.apache.mina.filter.codec.ProtocolDecoderOutput;  
9. **public** **class** MyProtocalDecoder **implements** ProtocolDecoder \{  
10.     **private** **final** AttributeKey CONTEXT = **new** AttributeKey(getClass(), "context");  
11.     **private** **final** Charset charset;  
12.     **private** **int** maxPackLength = 100;  
13.     **public** MyProtocalDecoder() \{  
14.         **this**(Charset.defaultCharset());  
15.     \}  
16.     **public** MyProtocalDecoder(Charset charset) \{  
17.         **this**.charset = charset;  
18.     \}  
19.     **public** **int** getMaxLineLength() \{  
20.         **return** maxPackLength;  
21.     \}  
22.     **public** **void** setMaxLineLength(**int** maxLineLength) \{  
23.         **if** (maxLineLength <= 0) \{  
24.             **throw** **new** IllegalArgumentException("maxLineLength: " + maxLineLength);  
25.         \}  
26.         **this**.maxPackLength = maxLineLength;  
27.     \}  
28.     **private** Context getContext(IoSession session) \{  
29.         Context ctx;  
30.         ctx = (Context) session.getAttribute(CONTEXT);  
31.         **if** (ctx == **null**) \{  
32.             ctx = **new** Context();  
33.             session.setAttribute(CONTEXT, ctx);   
34.         \}   
35.         **return** ctx;  
36.     \}  
37.     **public** **void** decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) **throws** Exception \{  
38.         **final** **int** packHeadLength = 5;  
39.         //先获取上次的处理上下文，其中可能有未处理完的数据  
40.         Context ctx = getContext(session);  
41.         // 先把当前buffer中的数据追加到Context的buffer当中   
42.         ctx.append(in);   
43.         //把position指向0位置，把limit指向原来的position位置  
44.         IoBuffer buf = ctx.getBuffer();  
45.         buf.flip();   
46.         // 然后按数据包的协议进行读取  
47.         **while** (buf.remaining() >= packHeadLength) \{  
48.             buf.mark();  
49.             // 读取消息头部分  
50.             **int** length = buf.getInt();  
51.             **byte** flag = buf.get();  
52.             //检查读取的包头是否正常，不正常的话清空buffer  
53.             **if** (length<0 ||length > maxPackLength) \{  
54.                 buf.clear();   
55.                 **break**;  
56.             \}   
57.             //读取正常的消息包，并写入输出流中，以便IoHandler进行处理  
58.             **else** **if** (length >= packHeadLength && length - packHeadLength <= buf.remaining()) \{  
59.                 **int** oldLimit2 = buf.limit();  
60.                 buf.limit(buf.position() + length - packHeadLength);  
61.                 String content = buf.getString(ctx.getDecoder());  
62.                 buf.limit(oldLimit2);  
63.                 MyProtocalPack pack = **new** MyProtocalPack(flag, content);  
64.                 out.write(pack);  
65.             \} **else** \{  
66.                 // 如果消息包不完整  
67.                 // 将指针重新移动消息头的起始位置   
68.                 buf.reset();   
69.                 **break**;  
70.             \}  
71.         \}  
72.         **if** (buf.hasRemaining()) \{  
73.             // 将数据移到buffer的最前面   
74.                 IoBuffer temp = IoBuffer.allocate(maxPackLength).setAutoExpand(**true**);  
75.                 temp.put(buf);  
76.                 temp.flip();  
77.                 buf.clear();  
78.                 buf.put(temp);  
79.                    
80.         \} **else** \{ // 如果数据已经处理完毕，进行清空  
81.             buf.clear();   
82.         \}  
83.           
84.           
85.     \}  
86.     **public** **void** finishDecode(IoSession session, ProtocolDecoderOutput out) **throws** Exception \{  
87.     \}  
88.     **public** **void** dispose(IoSession session) **throws** Exception \{   
89.         Context ctx = (Context) session.getAttribute(CONTEXT);  
90.         **if** (ctx != **null**) \{  
91.             session.removeAttribute(CONTEXT);  
92.         \}  
93.     \}  
94.     //记录上下文，因为数据触发没有规模，很可能只收到数据包的一半  
95.     //所以，需要上下文拼起来才能完整的处理  
96.     **private** **class** Context \{  
97.         **private** **final** CharsetDecoder decoder;  
98.         **private** IoBuffer buf;  
99.         **private** **int** matchCount = 0;  
100.         **private** **int** overflowPosition = 0;  
101.         **private** Context() \{  
102.             decoder = charset.newDecoder();  
103.             buf = IoBuffer.allocate(80).setAutoExpand(**true**);  
104.         \}  
105.         **public** CharsetDecoder getDecoder() \{  
106.             **return** decoder;  
107.         \}  
108.         **public** IoBuffer getBuffer() \{  
109.             **return** buf;  
110.         \}  
111.         **public** **int** getOverflowPosition() \{  
112.             **return** overflowPosition;  
113.         \}  
114.         **public** **int** getMatchCount() \{  
115.             **return** matchCount;  
116.         \}  
117.         **public** **void** setMatchCount(**int** matchCount) \{  
118.             **this**.matchCount = matchCount;  
119.         \}  
120.         **public** **void** reset() \{  
121.             overflowPosition = 0;  
122.             matchCount = 0;  
123.             decoder.reset();  
124.         \}  
125.         **public** **void** append(IoBuffer in) \{   
126.             getBuffer().put(in);  
127.             
128.         \}  
129.    
130.     \}  
131. \}

在MyProtocalServer中，添加自己实现的Log4jFilter和编解码过滤器：

**\[java\]**  [view plain][] [copy][view plain]

1.  **package** com.gftech.mytool.mina;  
2.  **import** java.io.IOException;  
3.  **import** java.net.InetSocketAddress;  
4.  **import** java.nio.charset.Charset;  
5.  **import** org.apache.log4j.Logger;  
6.  **import** org.apache.log4j.PropertyConfigurator;  
7.  **import** org.apache.mina.core.service.IoAcceptor;  
8.  **import** org.apache.mina.core.service.IoHandlerAdapter;  
9.  **import** org.apache.mina.core.session.IdleStatus;  
10. **import** org.apache.mina.core.session.IoSession;  
11. **import** org.apache.mina.filter.codec.ProtocolCodecFilter;  
12. **import** org.apache.mina.transport.socket.nio.NioSocketAcceptor;  
13. **public** **class** MyProtocalServer \{  
14.     **private** **static** **final** **int** PORT = 2500;  
15.     **static** Logger logger = Logger.getLogger(MyProtocalServer.**class**);  
16.     **public** **static** **void** main(String\[\] args) **throws** IOException \{  
17.         PropertyConfigurator.configure("conf//log4j.properties");  
18.         IoAcceptor acceptor = **new** NioSocketAcceptor();  
19.         Log4jFilter lf = **new** Log4jFilter(logger);  
20.         acceptor.getFilterChain().addLast("logger", lf);  
21.       
22.         acceptor.getFilterChain().addLast("codec", **new** ProtocolCodecFilter(**new** MyProtocalCodecFactory(Charset.forName("GBK"))));  
23.         acceptor.getSessionConfig().setReadBufferSize(1024);  
24.         acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH\_IDLE, 10);  
25.         acceptor.setHandler(**new** MyHandler());  
26.         acceptor.bind(**new** InetSocketAddress(PORT));  
27.         System.out.println("start server ...");  
28.     \}  
29. \}  
30. **class** MyHandler **extends** IoHandlerAdapter \{  
31.     **static** Logger logger = Logger.getLogger(MyHandler.**class**);  
32.     @Override  
33.     **public** **void** exceptionCaught(IoSession session, Throwable cause) **throws** Exception \{  
34.         cause.printStackTrace();  
35.     \}  
36.     @Override  
37.     **public** **void** messageReceived(IoSession session, Object message) **throws** Exception \{  
38.         MyProtocalPack pack=(MyProtocalPack)message;  
39.         logger.debug("Rec:" + pack);  
40.     \}  
41.     @Override  
42.     **public** **void** sessionIdle(IoSession session, IdleStatus status) **throws** Exception \{  
43.         logger.debug("IDLE " + session.getIdleCount(status));  
44.     \}  
45. \}

编写一个客户端程序进行测试：

**\[java\]**  [view plain][] [copy][view plain]

1. **package** com.gftech.mytool.mina;  
2. **import** java.io.DataOutputStream;  
3. **import** java.net.Socket;  
4. **public** **class** MyProtocalClient \{  
5.       
6.     **public** **static** **void** main(String\[\] args) \{  
7.         **try** \{  
8.             Socket socket = **new** Socket("127.0.0.1", 2500);  
9.             DataOutputStream out =  **new** DataOutputStream( socket.getOutputStream() ) ;  
10.             **for** (**int** i = 0; i < 1000; i++) \{  
11.                 MyProtocalPack pack=**new** MyProtocalPack((**byte**)i,i+"测试MyProtocalaaaaaaaaaaaaaa");  
12.                 out.writeInt(pack.getLength());  
13.                 out.write(pack.getFlag());  
14.                 out.write(pack.getContent().getBytes());  
15.                 out.flush();  
16.                 System.out.println(i + " sended");  
17.             \}  
18.             Thread.sleep(1000 );  
19.         \} **catch** (Exception e) \{  
20.             e.printStackTrace();  
21.         \}  
22.     \}  
23. \}

也可以用IoConnector实现自己的客户端：

**\[java\]**  [view plain][] [copy][view plain]

1. **package** com.gftech.mytool.mina;  
2. **import** java.io.IOException;  
3. **import** java.net.InetSocketAddress;  
4. **import** java.nio.charset.Charset;  
5. **import** org.apache.mina.core.future.ConnectFuture;  
6. **import** org.apache.mina.core.future.IoFutureListener;  
7. **import** org.apache.mina.core.service.IoConnector;  
8. **import** org.apache.mina.core.service.IoHandlerAdapter;  
9. **import** org.apache.mina.core.session.IdleStatus;  
10. **import** org.apache.mina.core.session.IoSession;  
11. **import** org.apache.mina.filter.codec.ProtocolCodecFilter;  
12. **import** org.apache.mina.transport.socket.nio.NioSocketConnector;  
13. **public** **class** MyProtocalClient2 \{  
14.     **private** **static** **final** String HOST = "192.168.10.8";  
15.     **private** **static** **final** **int** PORT = 2500;  
16.     **static** **long** counter = 0;  
17.     **final** **static** **int** FC1 = 100;  
18.     **static** **long** start = 0;  
19.     /\*\* 
20.      \* 使用Mina的框架结构进行测试 
21.      \*  
22.      \* @param args 
23.      \*/  
24.     **public** **static** **void** main(String\[\] args) **throws** IOException \{  
25.         start = System.currentTimeMillis();  
26.         IoConnector connector = **new** NioSocketConnector();  
27.         connector.getFilterChain().addLast("codec", **new** ProtocolCodecFilter(**new** MyProtocalCodecFactory(Charset.forName("GBK"))));  
28.         connector.setHandler(**new** TimeClientHandler2());  
29.         connector.getSessionConfig().setReadBufferSize(100);  
30.         connector.getSessionConfig().setIdleTime(IdleStatus.BOTH\_IDLE, 10);   
31.         ConnectFuture connFuture = connector.connect(**new** InetSocketAddress(HOST, PORT));  
32.         connFuture.addListener(**new** IoFutureListener<ConnectFuture>() \{  
33.             **public** **void** operationComplete(ConnectFuture future) \{  
34.                 **try** \{  
35.                     **if** (future.isConnected()) \{  
36.                         IoSession session = future.getSession();   
37.                         sendData(session);    
38.                     \} **else** \{  
39.                         System.out.println("连接不存在 ");  
40.                     \}  
41.                 \} **catch** (Exception e) \{  
42.                     e.printStackTrace();  
43.                 \}  
44.             \}  
45.         \});  
46.         System.out.println("start client ...");  
47.     \}  
48.     **public** **static** **void** sendData(IoSession session) **throws** IOException \{  
49.         **for** (**int** i = 0; i < FC1; i++) \{  
50.             String content = "afdjkdafk张新波测试" + i;  
51.             MyProtocalPack pack = **new** MyProtocalPack((**byte**) i, content);  
52.             session.write(pack);  
53.             System.out.println("send data:" + pack);  
54.         \}  
55.     \}  
56. \}  
57. **class** TimeClientHandler2 **extends** IoHandlerAdapter \{  
58.     @Override  
59.     **public** **void** sessionOpened(IoSession session) \{  
60.         // Set reader idle time to 10 seconds.  
61.         // sessionIdle(...) method will be invoked when no data is read  
62.         // for 10 seconds.  
63.         session.getConfig().setIdleTime(IdleStatus.READER\_IDLE, 60);  
64.     \}  
65.     @Override  
66.     **public** **void** sessionClosed(IoSession session) \{  
67.         // Print out total number of bytes read from the remote peer.  
68.         System.err.println("Total " + session.getReadBytes() + " byte(s)");  
69.     \}  
70.     @Override  
71.     **public** **void** sessionIdle(IoSession session, IdleStatus status) \{  
72.         // Close the connection if reader is idle.  
73.         **if** (status == IdleStatus.READER\_IDLE) \{  
74.             session.close(**true**);  
75.         \}  
76.     \}  
77.     @Override  
78.     **public** **void** messageReceived(IoSession session, Object message) \{  
79.         MyProtocalPack pack = (MyProtocalPack) message;  
80.         System.out.println("rec:" + pack);  
81.     \}  
82. \}

转载地址：

http://blog.csdn.net/sinboy/article/details/3988642

[view plain]: http://blog.csdn.net/sinboy/article/details/3988642#