netty-java开发的网络通信服务

package com.eatook.neatook; import java.io.BufferedReader; import java.io.InputStreamReader; import java.net.URL; import java.net.URLConnection; import java.nio.ByteOrder; import java.util.HashMap; import java.util.Map; import java.util.Timer; import java.util.TimerTask; import java.util.logging.Level; import java.util.logging.Logger; import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.buffer.ChannelBuffers; import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelEvent; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelStateEvent; import org.jboss.netty.channel.ExceptionEvent; import org.jboss.netty.channel.MessageEvent; import org.jboss.netty.channel.SimpleChannelHandler; import org.jboss.netty.channel.group.ChannelGroup; import org.jboss.netty.channel.group.DefaultChannelGroup; public class PadServerHandler extends SimpleChannelHandler { private static final Logger logger = Logger .getLogger(PadServerHandler.class.getName()); static final ChannelGroup channels = new DefaultChannelGroup(); static Map<String, Integer> padversionManager = new HashMap<String, Integer>(); static Map<String, Integer> padId = new HashMap<String, Integer>(); /** * 此构造方法主要用途： 启动一个timer来做数据库和客户端的版本检查和更新 */ public PadServerHandler() { Timer timer = new Timer(); if (padversionManager != null) { timer.schedule(new TimerTask() { @Override public void run() { try { /** * 1. 取得数据库的lastVersion 2. 遍历保存的clientList, 如果 * 某个ClientList.lastVersion < lastVersion 3. * 直接push一条新的信息. 客户端的返回处理在messageRecieved中。 4. * 如果不用更新，就直接检查客户端是否在线。如果超时，就报警。 * 数据包格式：长度+消息类型+版本号+ipad编号+处理流水号 */ byte[] msgBody = null; for (Channel c : channels) { // 获取连接上来的客户端的ip String ipport = c.getRemoteAddress().toString(); String ip = ipport.substring(0, ipport.length() - 6); // 获取客户端的版本并+1，为推送做准备 if (padversionManager.get(ip) != null) { int currentVersion = padversionManager.get(ip) + 1; // 获取客户端当前版本的下一版的json String json = new VersionChecker() .VersionCheck(ip); // 判断是否为空 if (json != null) { // 构造body msgBody = json.getBytes(); } // 算长度 int msgLen = msgBody.length + 16; // 判断是否已经是最新版本，如果不是则推送给客户端 if (!"no new version".equals(json)) { ChannelBuffer buf = ChannelBuffers.buffer( ByteOrder.LITTLE_ENDIAN, msgLen); buf.writeInt(msgLen);// 长度 buf.writeInt(MsgTypeEnum.CHECKVERSION);// 消息类型 buf.writeInt(currentVersion);// 版本号 buf.writeInt(0);// ipad编号 buf.writeBytes(msgBody);// 主要数据 c.write(buf); } } } } catch (Exception e) { e.printStackTrace(); } } }, 1000, 5000);// 程序启动1秒后开始执行，每次间隔5秒 } } @Override public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception { if (e instanceof ChannelStateEvent) { logger.info(e.toString()); } super.handleUpstream(ctx, e); } @Override public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { logger.info(e.toString()); channels.add(ctx.getChannel()); } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { // Unregister the channel from the global channel list // so the channel does not receive messages anymore. /** * 客户端失去连接报警,并做数据库记录处理 */ String ipport = e.getChannel().getRemoteAddress().toString(); String ip = ipport.substring(0, ipport.length() - 6); // TODO 报警 System.out.println(e.getChannel().getRemoteAddress() + ": miss connection"); // 连接url String url = "http://192.168.1.109/eatook-api/jsondevice.php?function" + "=addMobileDeviceError&md_id="// ipadID + 1 + "&description="// 描述 + 2 + "&user_password=" + "a"; URL U = new URL(url); URLConnection connection = U.openConnection(); connection.connect(); // 获得结果 BufferedReader in = new BufferedReader(new InputStreamReader( connection.getInputStream())); String line = ""; String result = ""; while ((line = in.readLine()) != null) { result += line; } in.close(); // TODO 数据调用结果 System.out.println(result); // 输出 channels.remove(e.getChannel()); padversionManager.remove(ip); padId.remove(ip); } @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) { // Convert to a String first. Object o = e.getMessage(); ChannelBuffer m = null; // 用ip构造HashMap的key String ipport = e.getRemoteAddress().toString(); String ip = ipport.substring(0, ipport.length() - 6); if (o instanceof ChannelBuffer) { m = (ChannelBuffer) o; } if (m != null) { @SuppressWarnings("unused") int len = m.readInt(); int msgType = m.readInt(); int lastVersion = m.readInt(); int ipadId = m.readInt(); // 如果是版本check的回复：将内存中对应的版本号更新 if (msgType == MsgTypeEnum.RESP_CHECKVERSION) { padversionManager.remove(ip); padversionManager.put(ip, lastVersion); System.out.println(ip + ":" + lastVersion); } // 接受开始时客户端传来的当前客户端版本 if (msgType == MsgTypeEnum.CLIENTVERSION) { // 记录连接上来的ipad的版本号 padversionManager.put(ip, lastVersion); // 记录连接上来的ipad的编号 padId.put(ip, ipadId); // 回写数据，保证连接畅通 e.getChannel().write(m); System.out.println(ip + ":" + lastVersion); } } System.out.println(e.getMessage()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) { logger.log(Level.WARNING, "Unexpected exception from downstream.", e.getCause()); e.getChannel().close(); } }