版权声明：本文为原创文章，转载请声明http://www.cnblogs.com/unityExplorer/p/6986474.html
上一篇讲到了数据的处理，这一篇主要讲使用多线程收发消息
//创建消息数据模型
//正式项目中，消息的结构一般是消息长度+消息id+消息主体内容
public class Message
{
public IExtensible protobuf;
public int messageId;
} public class SocketClientTemp : MonoBehaviour
{
const int packageMaxLength = ; Socket mSocket;
Thread threadSend;
Thread threadRecive;
Queue<Message> allMessages = new Queue<Message>();
Queue<byte[]> sendQueue = new Queue<byte[]>(); public bool Init()
{
//创建一个socket对象
mSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
return SocketConnection("此处是ip", );
} void Update()
{
AnalysisMessage();
} /// <summary>
/// 建立服务器连接
/// </summary>
/// <param name="ip">服务器的ip地址</param>
/// <param name="port">端口</param>
bool SocketConnection(string ip, int port)
{
try
{
IPEndPoint ipep = new IPEndPoint(IPAddress.Parse(ip), port);
//同步连接服务器，实际使用时推荐使用异步连接，处理方式会在下一篇讲断线重连时讲到
mSocket.Connect(ipep);
//连接成功后，创建两个线程，分别用于发送和接收消息
threadSend = new Thread(new ThreadStart(SendMessage));
threadSend.Start();
threadRecive = new Thread(new ThreadStart(ReceiveMessage));
threadRecive.Start();
return true;
}
catch (Exception e)
{
Debug.Log(e.ToString());
Close();
return false;
}
} #region ...发送消息
/// <summary>
/// 添加数据到发送队列
/// </summary>
/// <param name="protobufModel"></param>
/// <param name="messageId"></param>
public void AddSendMessageQueue(IExtensible protobufModel, int messageId)
{
sendQueue.Enqueue(BuildPackage(protobufModel, messageId));
} void SendMessage()
{
//循环获取发送队列中第一个数据，然后发送到服务器
while (true)
{
if (sendQueue.Count == )
{
Thread.Sleep();
continue;
}
if (!mSocket.Connected)
{
Close();
break;
}
else
Send(sendQueue.Peek());//发送队列中第一条数据
}
} void Send(byte[] bytes)
{
try
{
mSocket.Send(bytes, SocketFlags.None);
//发送成功后，从发送队列中移除已发送的消息
sendQueue.Dequeue();
}
catch (SocketException e)
{
//如果错误码为10035，说明服务器缓存区满了，所以等100毫秒再次发送
if (e.NativeErrorCode == )
{
Thread.Sleep();
Send(bytes);
}
else
Debug.Log(e.ToString());
}
}
#endregion #region ...接收消息
/// <summary>
/// 解析收到的消息
/// </summary>
void AnalysisMessage()
{
while (allMessages.Count > )
{
int id = allMessages.Dequeue().messageId;
switch (id)
{
//根据消息id做不同的处理
}
}
} /// <summary>
/// 接收数据
/// </summary>
void ReceiveMessage()
{
while (true)
{
if (!mSocket.Connected)
break;
byte[] recvBytesHead = GetBytesReceive();
int bodyLength = IPAddress.NetworkToHostOrder(BitConverter.ToInt32(recvBytesHead, ));
byte[] recvBytesBody = GetBytesReceive(bodyLength); byte[] messageId = new byte[];
Array.Copy(recvBytesBody, , messageId, , );
byte[] messageBody = new byte[bodyLength - ];
Array.Copy(recvBytesBody, , messageBody, , bodyLength - ); if (BitConverter.IsLittleEndian)
Array.Reverse(messageId);
FillAllPackages(BitConverter.ToInt32(messageId, ), messageBody);
}
} /// <summary>
/// 填充接收消息队列
/// </summary>
/// <param name="messageId"></param>
/// <param name="messageBody"></param>
void FillAllPackages(int messageId, byte[] messageBody)
{
switch (messageId)
{
//根据消息id处理消息，并添加到接收消息队列
case :
allMessages.Enqueue(new Message()
{
protobuf = ProtobufSerilizer.DeSerialize<TestTemp>(messageBody),
messageId = messageId
});
break;
}
} /// <summary>
/// 接收数据并处理
/// </summary>
/// <param name="length"></param>
/// <returns></returns>
byte[] GetBytesReceive(int length)
{
byte[] recvBytes = new byte[length];
while (length > )
{
byte[] receiveBytes = new byte[length < packageMaxLength ? length : packageMaxLength];
int iBytesBody = ;
if (length >= receiveBytes.Length)
iBytesBody = mSocket.Receive(receiveBytes, receiveBytes.Length, );
else
iBytesBody = mSocket.Receive(receiveBytes, length, );
receiveBytes.CopyTo(recvBytes, recvBytes.Length - length);
length -= iBytesBody;
}
return recvBytes;
}
#endregion /// <summary>
/// 构建消息数据包
/// </summary>
/// <param name="protobufModel"></param>
/// <param name="messageId"></param>
byte[] BuildPackage(IExtensible protobufModel, int messageId)
{
byte[] b;
if (protobufModel != null)
b = ProtobufSerilizer.Serialize(protobufModel);
else
b = new byte[];
//消息长度(int数据，长度4) + 消息id(int数据，长度4) + 消息主体内容
ByteBuffer buf = ByteBuffer.Allocate(b.Length + + );
//消息长度 = 消息主体内容长度 + 消息id长度
buf.WriteInt(b.Length + );
buf.WriteInt(messageId); if (protobufModel != null)
buf.WriteBytes(b);
return buf.GetBytes();
} void OnDestroy()
{
//停止运行后，如果不关闭socket多线程，再次运行时，unity会卡死
Close();
} /// <summary>
/// 关闭socket,终止线程
/// </summary>
public void Close()
{
if (mSocket != null)
{
//微软官方推荐在关闭socket前先shutdown，但是经过测试，发现网络断开后，shutdown会无法执行
if (mSocket.Connected)
mSocket.Shutdown(SocketShutdown.Both);
mSocket.Close();
mSocket = null;
}
//关闭线程
if (threadSend != null)
threadSend.Abort();
if (threadRecive != null)
threadRecive.Abort();
threadSend = null;
threadRecive = null;
}
}
到这里，使用socket处理消息的收发就基本结束了，但是，某些项目为了增强体验，可能还会增加断线重连的功能，这个功能会在下一篇讲到
