java netty socket庫和自定義C#socket庫利用protobuf進行通訊完整例項
阿新 • • 發佈：2018-12-30
之前的文章講述了socket通訊的一些基本知識，已經本人自定義的C#版本的socket、和java netty 庫的二次封裝，但是沒有真正的發表測試用例。
本文只是為了講解利用protobuf 進行C# 和 java的通訊。以及完整的例項程式碼
java 程式碼 svn 地址，本人開發工具是NetBeans 8.0.2 使用 maven 專案編譯
http://code.taobao.org/svn/flynetwork_csharp/trunk/BlogTest
c# 程式碼 svn 地址 使用的是 vs 2013 .net 4.5
http://code.taobao.org/svn/flynetwork_csharp/trunk/Flynetwork/BlogTest
編譯工具下載
http://files.cnblogs.com/files/ty408/Sz.ExcelTools.zip
本文著重以C# socket作為伺服器端，java netty作為socket的客戶端進行訪問通訊
首先附上proto的message檔案
package Sz.Test.ProtoMessage;
//登陸訊息
message TestMessage {
//訊息列舉
enum Proto_Login {
ResTip = 101201;//伺服器推送提示
ReqLogin = 101102;//客戶端申請登陸
ReqChat = 101103;//客戶端申請聊天訊息
ResChat = 101203;//伺服器推送聊天訊息
}
//伺服器推送提示 ResTip
message ResTipMessage {
required string msg = 1;//提示內容
}
//客戶端申請登陸 ReqLogin
message ReqLoginMessage {
required string userName = 1;//登陸使用者名稱
required string userPwd = 2;//登陸密碼
}
//客戶端申請登陸 ReqChat
message ReqChatMessage {
required string msg = 1;//提示內容
}
//客戶端申請登陸 ResChat
message ResChatMessage {
required string msg = 1;//提示內容
}
}本人編譯工具自帶生產訊息，和對應的handler
先把proto檔案編譯生產後，放到哪裡，然後建立伺服器監聽程式碼
上一篇文章講到由於java和C#預設網路端緒不一樣，java是標準端緒大端序，C#使用的小端序。
1 MarshalEndian.JN = MarshalEndian.JavaOrNet.Java; 2 Sz.Network.SocketPool.ListenersBox.Instance.SetParams(new MessagePool(), typeof(MarshalEndian)); 3 Sz.Network.SocketPool.ListenersBox.Instance.Start("tcp:*:9527");
所以在我開啟伺服器監聽的時候設定解碼器和編碼器的解析風格為java
然後建立一個檔案chat資料夾用於存放handler檔案就是剛才工具生成 目錄下的 ExcelSource\protobuf\net\Handler
這一系列檔案
1 if (message.MsgID == (int)Sz.Test.ProtoMessage.TestMessage.Proto_Login.ReqLogin) 2 { 3 //構建訊息 4 Sz.Test.ProtoMessage.TestMessage.ReqLoginMessage loginmessage = new Test.ProtoMessage.TestMessage.ReqLoginMessage(); 5 object msg = DeSerialize(message.MsgBuffer, loginmessage); 6 //構建handler 7 Test.ProtoMessage.ReqLoginHandler handler = new Test.ProtoMessage.ReqLoginHandler(); 8 handler.Session = client; 9 handler.Message = loginmessage; 10 //把handler交給 登入 執行緒處理 11 ThreadManager.Instance.AddTask(ServerManager.LoginThreadID, handler); 12 } 13 else if (message.MsgID == (int)Sz.Test.ProtoMessage.TestMessage.Proto_Login.ReqChat) 14 { 15 //構建訊息 16 Sz.Test.ProtoMessage.TestMessage.ReqChatMessage loginmessage = new Test.ProtoMessage.TestMessage.ReqChatMessage(); 17 object msg = DeSerialize(message.MsgBuffer, loginmessage); 18 //構建handler 19 Test.ProtoMessage.ReqChatHandler handler = new Test.ProtoMessage.ReqChatHandler(); 20 handler.Session = client; 21 handler.Message = loginmessage; 22 //把handler交給 聊天 執行緒處理 23 ThreadManager.Instance.AddTask(ServerManager.ChatThreadID, handler); 24 }
收到訊息後的處理判斷傳入的訊息id是什麼型別的，然後對應反序列化byte[]陣列為訊息
最後把訊息和生成handler移交到對應的執行緒處理
登入的訊息全部交給 LoginThread 執行緒 去處理 ，這樣在真實的執行環境下，能保證單點登入問題；
聊天訊息全部交給 ChatThread 執行緒 去處理 這樣的好處是，聊天與登入無關；
收到登入訊息的處理
1 public class ReqLoginHandler : TcpHandler 2 { 3 public override void Run() 4 { 5 6 var message = (Sz.Test.ProtoMessage.TestMessage.ReqLoginMessage)this.Message; 7 Sz.Test.ProtoMessage.TestMessage.ResTipMessage tip = new TestMessage.ResTipMessage(); 8 if (message.userName == "admin" && message.userPwd == "admin") 9 { 10 Logger.Debug("收到登入訊息 登入完成"); 11 tip.msg = "登入完成"; 12 } 13 else 14 { 15 Logger.Debug("收到登入訊息 使用者名稱或者密碼錯誤"); 16 tip.msg = "使用者名稱或者密碼錯誤"; 17 } 18 byte[] buffer = MessagePool.Serialize(tip); 19 this.Session.SendMsg(new Network.SocketPool.SocketMessage((int)Sz.Test.ProtoMessage.TestMessage.Proto_Login.ResTip, buffer)); 20 } 21 }
收到聊天訊息的處理
1 public class ReqChatHandler : TcpHandler 2 { 3 public override void Run() 4 { 5 var message = (Sz.Test.ProtoMessage.TestMessage.ReqChatMessage)this.Message; 6 Logger.Debug("收到來自客戶端聊天訊息：" + message.msg); 7 Sz.Test.ProtoMessage.TestMessage.ResChatMessage chat = new TestMessage.ResChatMessage(); 8 chat.msg = "伺服器廣播：" + message.msg; 9 byte[] buffer = MessagePool.Serialize(chat); 10 this.Session.SendMsg(new Network.SocketPool.SocketMessage((int)Sz.Test.ProtoMessage.TestMessage.Proto_Login.ResChat, buffer)); 11 } 12 }
接下來我們構建
java版本基於netty 二次封裝的socket客戶端
1 package sz.network.socketpool.nettypool; 2 3 import Sz.Test.ProtoMessage.Test.TestMessage; 4 import com.google.protobuf.InvalidProtocolBufferException; 5 import io.netty.channel.ChannelHandlerContext; 6 import java.io.BufferedReader; 7 import java.io.IOException; 8 import java.io.InputStreamReader; 9 import java.util.logging.Level; 10 import org.apache.log4j.Logger; 11 12 /** 13 * 14 * @author Administrator 15 */ 16 public class TestClient { 17 18 static final Logger log = Logger.getLogger(TestClient.class); 19 static NettyTcpClient client = null; 20 21 public static void main(String[] args) { 22 client = new NettyTcpClient("127.0.0.1", 9527, true, new NettyMessageHandler() { 23 24 @Override 25 public void channelActive(ChannelHandlerContext session) { 26 log.info("連線伺服器成功："); 27 //構建錯誤的登入訊息 28 TestMessage.ReqLoginMessage.Builder newBuilder = TestMessage.ReqLoginMessage.newBuilder(); 29 newBuilder.setUserName("a"); 30 newBuilder.setUserPwd("a"); 31 //傳送訊息 32 TestClient.client.sendMsg(new NettyMessageBean(TestMessage.Proto_Login.ReqLogin_VALUE, newBuilder.build().toByteArray())); 33 34 //構建正確的登入訊息 35 TestMessage.ReqLoginMessage.Builder newBuilder1 = TestMessage.ReqLoginMessage.newBuilder(); 36 newBuilder1.setUserName("admin"); 37 newBuilder1.setUserPwd("admin"); 38 TestClient.client.sendMsg(new NettyMessageBean(TestMessage.Proto_Login.ReqLogin_VALUE, newBuilder1.build().toByteArray())); 39 } 40 41 @Override 42 public void readMessage(NettyMessageBean msg) { 43 try { 44 if (msg.getMsgid() == TestMessage.Proto_Login.ResTip_VALUE) { 45 TestMessage.ResTipMessage tipmessage = TestMessage.ResTipMessage.parseFrom(msg.getMsgbuffer()); 46 log.info("收到提示資訊：" + tipmessage.getMsg()); 47 } else if (msg.getMsgid() == TestMessage.Proto_Login.ResChat_VALUE) { 48 TestMessage.ResChatMessage tipmessage = TestMessage.ResChatMessage.parseFrom(msg.getMsgbuffer()); 49 log.info("收到聊天訊息：" + tipmessage.getMsg()); 50 } 51 } catch (InvalidProtocolBufferException ex) { 52 log.error("收到訊息：" + msg.getMsgid() + " 解析出錯：" + ex); 53 } 54 } 55 56 @Override 57 public void closeSession(ChannelHandlerContext session) { 58 log.info("連線關閉或者連線不成功："); 59 } 60 61 @Override 62 public void exceptionCaught(ChannelHandlerContext session, Throwable cause) { 63 log.info("錯誤：" + cause.toString()); 64 } 65 }); 66 client.Connect(); 67 68 BufferedReader strin = new BufferedReader(new InputStreamReader(System.in)); 69 while (true) { 70 try { 71 String str = strin.readLine(); 72 //構建聊天訊息 73 TestMessage.ReqChatMessage.Builder chatmessage = TestMessage.ReqChatMessage.newBuilder(); 74 chatmessage.setMsg(str); 75 TestClient.client.sendMsg(new NettyMessageBean(TestMessage.Proto_Login.ReqChat_VALUE, chatmessage.build().toByteArray())); 76 } catch (IOException ex) { 77 } 78 } 79 80 } 81 82 }
接下來我們看看效果
我設定了斷線重連功能，我們來測試一下，把伺服器關閉
可以看到沒3秒向伺服器發起一次請求；
知道伺服器再次開啟連結成功
完整的通訊示例演示就完了；
程式碼我不在上傳了，請各位使用svn下載好麼？？？？
需要注意的是，訊息的解碼器和編碼器，一定要雙方都遵守你自己的契約。比如我在編碼訊息格式的時候先寫入訊息包的長度，然後跟上訊息的id，再是訊息的內容
所以解碼的時候，先讀取一個訊息長度，在讀取一個訊息id，如果本次收到的訊息位元組數不夠長度那麼留存起來以用於下一次收到位元組陣列追加後再一起解析。
這樣就能解決粘包的問題。
附上C#版本的解析器
1 using System; 2 using System.Collections.Generic; 3 using System.IO; 4 using System.Linq; 5 using System.Text; 6 using System.Threading.Tasks; 7 8 /** 9 * 10 * @author 失足程式設計師 11 * @Blog http://www.cnblogs.com/ty408/ 12 * @mail [email protected] 13 * @phone 13882122019 14 * 15 */ 16 namespace Sz.Network.SocketPool 17 { 18 public class MarshalEndian : IMarshalEndian 19 { 20 21 public enum JavaOrNet 22 { 23 Java, 24 Net, 25 } 26 27 public MarshalEndian() 28 { 29 30 } 31 32 public static JavaOrNet JN = JavaOrNet.Net; 33 34 /// <summary> 35 /// 讀取大端序的int 36 /// </summary> 37 /// <param name="value"></param> 38 public int ReadInt(byte[] intbytes) 39 { 40 Array.Reverse(intbytes); 41 return BitConverter.ToInt32(intbytes, 0); 42 } 43 44 /// <summary> 45 /// 寫入大端序的int 46 /// </summary> 47 /// <param name="value"></param> 48 public byte[] WriterInt(int value) 49 { 50 byte[] bs = BitConverter.GetBytes(value); 51 Array.Reverse(bs); 52 return bs; 53 } 54 55 //用於儲存剩餘未解析的位元組數 56 private List<byte> _LBuff = new List<byte>(2); 57 58 //位元組數常量一個訊息id4個位元組 59 const long ConstLenght = 4L; 60 61 public void Dispose() 62 { 63 this.Dispose(true); 64 GC.SuppressFinalize(this); 65 } 66 67 protected virtual void Dispose(bool flag1) 68 { 69 if (flag1) 70 { 71 IDisposable disposable = this._LBuff as IDisposable; 72 if (disposable != null) { disposable.Dispose(); } 73 } 74 } 75 76 public byte[] Encoder(SocketMessage msg) 77 { 78 MemoryStream ms = new MemoryStream(); 79 BinaryWriter bw = new BinaryWriter(ms, UTF8Encoding.Default); 80 byte[] msgBuffer = msg.MsgBuffer; 81 82 if (msgBuffer != null) 83 { 84 switch (JN) 85 { 86 case JavaOrNet.Java: 87 bw.Write(WriterInt(msgBuffer.Length + 4)); 88 bw.Write(WriterInt(msg.MsgID)); 89 break; 90 case JavaOrNet.Net: 91 bw.Write((Int32)(msgBuffer.Length + 4)); 92 bw.Write(msg.MsgID); 93 break; 94 } 95 96 bw.Write(msgBuffer); 97 } 98 else 99 { 100 switch (JN) 101 { 102 case JavaOrNet.Java: 103 bw.Write(WriterInt(0)); 104 break; 105 case JavaOrNet.Net: 106 bw.Write((Int32)0); 107 break; 108 } 109 } 110 bw.Close(); 111 ms.Close(); 112 bw.Dispose(); 113 ms.Dispose(); 114 return ms.ToArray(); 115 } 116 117 public List<SocketMessage> Decoder(byte[] buff, int len) 118 { 119 //拷貝本次的有效位元組 120 byte[] _b = new byte[len]; 121 Array.Copy(buff, 0, _b, 0, _b.Length); 122 buff = _b; 123 if (this._LBuff.Count > 0) 124 { 125 //拷貝之前遺留的位元組 126 this._LBuff.AddRange(_b); 127 buff = this._LBuff.ToArray(); 128 this._LBuff.Clear(); 129 this._LBuff = new List<byte>(2); 130 } 131 List<SocketMessage> list = new List<SocketMessage>(); 132 MemoryStream ms = new MemoryStream(buff); 133 BinaryReader buffers = new BinaryReader(ms, UTF8Encoding.Default); 134 try 135 { 136 byte[] _buff; 137 Label_0073: 138 //判斷本次解析的位元組是否滿足常量位元組數 139 if ((buffers.BaseStream.Length - buffers.BaseStream.Position) < ConstLenght) 140 { 141 _buff = buffers.ReadBytes((int)(buffers.BaseStream.Length - buffers.BaseStream.Position)); 142 this._LBuff.AddRange(_buff); 143 } 144 else 145 { 146 long offset = 0; 147 switch (JN) 148 { 149 case JavaOrNet.Java: 150 offset = ReadInt(buffers.ReadBytes(4)); 151 break; 152 case JavaOrNet.Net: 153 offset = buffers.ReadInt32(); 154 break; 155 } 156 157 //剩餘位元組數大於本次需要讀取的位元組數 158 if (offset <= (buffers.BaseStream.Length - buffers.BaseStream.Position)) 159 { 160 int msgID = 0; 161 switch (JN) 162 { 163 case JavaOrNet.Java: 164 msgID = ReadInt(buffers.ReadBytes(4)); 165 break; 166 case JavaOrNet.Net: 167 msgID = buffers.ReadInt32(); 168 break; 169 } 170 _buff = buffers.ReadBytes((int)(offset - 4)); 171 list.Add(new SocketMessage(msgID, _buff)); 172 goto Label_0073; 173 } 174 else 175 { 176 //剩餘位元組數剛好小於本次讀取的位元組數 存起來，等待接受剩餘位元組數一起解析 177 buffers.BaseStream.Seek(ConstLenght, SeekOrigin.Current); 178 _buff = buffers.ReadBytes((int)(buffers.BaseStream.Length - buffers.BaseStream.Position)); 179 this._LBuff.AddRange(_buff); 180 } 181 } 182 } 183 catch { } 184 finally 185 { 186 buffers.Close(); 187 if (buffers != null) { buffers.Dispose(); } 188 ms.Close(); 189 if (ms != null) { ms.Dispose(); } 190 } 191 return list; 192 } 193 } 194 }
謝謝觀賞~！