Luk*_*ser 2 c# client multithreading chat server
我正在寻找一种处理断开连接的方法,因为每次关闭客户端时,服务器都会停止工作.我收到一条错误消息,在此行中"无法读取超出流的末尾":
string message = reader.ReadString();
此外,我需要一种方法来从客户端列表中删除断开连接的客户端.这是我的代码:服务器
using System;
using System.Threading;
using System.Net.Sockets;
using System.IO;
using System.Net;
using System.Collections.Generic;
namespace Server
{
class Server
{
public static List<TcpClient> clients = new List<TcpClient>();
static void Main(string[] args)
{
IPAddress ip = IPAddress.Parse("127.0.0.1");
TcpListener ServerSocket = new TcpListener(ip, 14000);
ServerSocket.Start();
Console.WriteLine("Server started.");
while (true)
{
TcpClient clientSocket = ServerSocket.AcceptTcpClient();
clients.Add(clientSocket);
handleClient client = new handleClient();
client.startClient(clientSocket);
}
}
}
public class handleClient
{
TcpClient clientSocket;
public void startClient(TcpClient inClientSocket)
{
this.clientSocket = inClientSocket;
Thread ctThread = new Thread(Chat);
ctThread.Start();
}
private void Chat()
{
while (true)
{
BinaryReader reader = new BinaryReader(clientSocket.GetStream());
while (true)
{
string message = reader.ReadString();
foreach (var client in Server.clients)
{
BinaryWriter writer = new BinaryWriter(client.GetStream());
writer.Write(message);
}
}
}
}
}
}
客户
using System;
using System.Net.Sockets;
using System.IO;
using System.Threading;
namespace Client
{
class Client
{
public static void Write()
{
TcpClient client = new TcpClient("127.0.0.1", 14000);
while (true)
{
string str = Console.ReadLine();
BinaryWriter writer = new BinaryWriter(client.GetStream());
writer.Write(str);
}
}
public static void Read()
{
TcpClient client = new TcpClient("127.0.0.1", 14000);
while (true)
{
BinaryReader reader = new BinaryReader(client.GetStream());
Console.WriteLine(reader.ReadString());
}
}
static void Main(string[] args)
{
Thread Thread = new Thread(Write);
Thread Thread2 = new Thread(Read);
Thread.Start();
Thread2.Start();
}
}
}
每次关闭客户端时,服务器都会停止工作.我收到一条错误消息,它"无法读取超出流的末尾"
从某种意义上说,这是完全正常的.也就是说,在使用时BinaryReader,它的正常行为是EndOfStreamException在到达流末尾时抛出.
为什么它会到达流的末尾？好吧,因为客户端断开连接,这就是流发生的事情.在套接字级别,真正发生的是读取操作以0作为读取的字节数完成.这表明客户端已正常关闭套接字,并且不会再发送任何数据.
在.NET API,这是翻译成的结束NetworkStream是TcpClient用来包裹Socket这实际上处理网络I/O对象.而这个NetworkStream对象又被你的BinaryReader对象包裹着.并且BinaryReader当它到达流的末尾时抛出该异常.
请注意,您的代码实际上并没有为用户提供关闭客户端的优雅方式.他们将不得不使用Ctrl+ C,或彻底杀死进程.使用前者具有执行正常关闭套接字的偶然效果,但这只是因为.NET正在处理进程的终止并在对象上运行终结器,例如TcpClient用于连接到服务器的对象和终结器调用Socket.Shutdown()告诉服务器它正在关闭.
如果你要杀死进程(例如使用任务管理器),你会发现IOException被抛出了.好的网络代码应该随时准备好看IOException; 网络不可靠,确实发生了故障.您希望做一些合理的操作,例如从连接中删除远程端点,而不是让整个程序崩溃.
现在,所有这一切,只是因为它EndOfStreamException是"正常的",这并不意味着您发布的代码,或者无论如何都是正确的网络编程方式的一个例子.你有很多问题:
TcpClient并且NetworkStream不要直接公开它,但是你可以使用该TcpClient.Client属性来让Socket对象做一个更好的优雅闭包,即一个端点可以指示它已经完成发送,并且仍然可以等到另一个端点也完成发送.TcpClient.Close()断开连接的方法就像挂断电话而不说"再见".运用Socket.Shutdown() is like finishing a phone call with a polite "okay, that's everything I wanted to say…was there anything else?"BinaryReader but not handling the EndOfStreamException correctly.Socket object, which supports full-duplex communications. There is no need to create a second connection just to do both reading and writing. A single connection suffices, and is better because when you split the send and receive into two connections, then you also need to add something to your protocol so that the server knows those two connections represent a single client (which your code does not actually do).Chat()方法中有一个额外的"while(true)".我已经修改了您的原始示例以解决上述所有问题,我在此处介绍过:
Server Program.cs:
class Program
{
private static readonly object _lock = new object();
private static readonly List<TcpClient> clients = new List<TcpClient>();
public static TcpClient[] GetClients()
{
lock (_lock) return clients.ToArray();
}
public static int GetClientCount()
{
lock (_lock) return clients.Count;
}
public static void RemoveClient(TcpClient client)
{
lock (_lock) clients.Remove(client);
}
static void Main(string[] args)
{
IPAddress ip = IPAddress.Parse("127.0.0.1");
TcpListener ServerSocket = new TcpListener(ip, 14000);
ServerSocket.Start();
Console.WriteLine("Server started.");
while (true)
{
TcpClient clientSocket = ServerSocket.AcceptTcpClient();
Console.WriteLine($"client connected: {clientSocket.Client.RemoteEndPoint}");
lock (_lock) clients.Add(clientSocket);
handleClient client = new handleClient();
client.startClient(clientSocket);
Console.WriteLine($"{GetClientCount()} clients connected");
}
}
}
服务器handleClient.cs:
public class handleClient
{
TcpClient clientSocket;
public void startClient(TcpClient inClientSocket)
{
this.clientSocket = inClientSocket;
Thread ctThread = new Thread(Chat);
ctThread.Start();
}
private void Chat()
{
BinaryReader reader = new BinaryReader(clientSocket.GetStream());
try
{
while (true)
{
string message = reader.ReadString();
foreach (var client in Program.GetClients())
{
BinaryWriter writer = new BinaryWriter(client.GetStream());
writer.Write(message);
}
}
}
catch (EndOfStreamException)
{
Console.WriteLine($"client disconnecting: {clientSocket.Client.RemoteEndPoint}");
clientSocket.Client.Shutdown(SocketShutdown.Both);
}
catch (IOException e)
{
Console.WriteLine($"IOException reading from {clientSocket.Client.RemoteEndPoint}: {e.Message}");
}
clientSocket.Close();
Program.RemoveClient(clientSocket);
Console.WriteLine($"{Program.GetClientCount()} clients connected");
}
}
客户计划.s:
class Program
{
private static readonly object _lock = new object();
private static bool _closed;
public static void Write(TcpClient client)
{
try
{
string str;
SocketShutdown reason = SocketShutdown.Send;
while ((str = Console.ReadLine()) != "")
{
lock (_lock)
{
BinaryWriter writer = new BinaryWriter(client.GetStream());
writer.Write(str);
if (_closed)
{
// Remote endpoint already said they are done sending,
// so we're done with both sending and receiving.
reason = SocketShutdown.Both;
break;
}
}
}
client.Client.Shutdown(reason);
}
catch (IOException e)
{
Console.WriteLine($"IOException writing to socket: {e.Message}");
}
}
public static void Read(TcpClient client)
{
try
{
while (true)
{
try
{
BinaryReader reader = new BinaryReader(client.GetStream());
Console.WriteLine(reader.ReadString());
}
catch (EndOfStreamException)
{
lock (_lock)
{
_closed = true;
return;
}
}
}
}
catch (IOException e)
{
Console.WriteLine($"IOException reading from socket: {e.Message}");
}
}
static void Main(string[] args)
{
TcpClient client = new TcpClient("127.0.0.1", 14000);
Thread writeThread = new Thread(() => Write(client));
Thread readThread = new Thread(() => Read(client));
writeThread.Start();
readThread.Start();
writeThread.Join();
readThread.Join();
client.Close();
Console.WriteLine("client exiting");
}
}
请注意,在大多数情况下,我没有解决您在代码中使用的不一致和非常规命名.唯一的例外是客户端代码中的线程变量,因为我真的不喜欢与类型名称完全匹配的大写局部变量.
您还有其他一些问题,上面的代码修订版没有解决这些问题.这些包括:
BinaryReader.这在很多方面都是令人烦恼的课程.我建议,特别是对于你只是处理文本的聊天服务器场景,你切换到使用StreamReader/ StreamWriter.Program类具有服务器代码,服务器代码知道Program该类.将服务器和客户端实现封装到它们自己的类中,与程序的主要入口点分开,并进一步将顶级服务器代码与每个客户端数据结构分离(使用C#event到允许顶级服务器代码被通知重要事件,例如需要从列表中删除客户端,而不必让每个客户端数据结构实际知道顶级服务器对象,更别关注它的客户端列表).通常情况下,我会说这些都超出了这样的答案范围,这已经很长了.我已经解决了你的代码中的直接问题,然后是一些,这在名义上是足够的.
但是,我一直想写几年前我写的基本网络编程示例的更新版本,作为一种"中间"示例,添加多个客户端支持,异步操作和使用最新的C#功能(喜欢async/ await).所以,我继续前进并花了一些时间来做到这一点.我想我最终会把它发布到我的博客上...这是另一个项目.与此同时,这是代码(请注意,这是一个完全从头开始的例子......这样做比尝试重新编写代码更有意义)...
Most of the grunt work for this implementation is in a single class shared by the server and client:
/// <summary>
/// Represents a remote end-point for the chat server and clients
/// </summary>
public sealed class ConnectedEndPoint : IDisposable
{
private readonly object _lock = new object();
private readonly Socket _socket;
private readonly StreamReader _reader;
private readonly StreamWriter _writer;
private bool _closing;
/// <summary>
/// Gets the address of the connected remote end-point
/// </summary>
public IPEndPoint RemoteEndPoint { get { return (IPEndPoint)_socket.RemoteEndPoint; } }
/// <summary>
/// Gets a <see cref="Task"/> representing the on-going read operation of the connection
/// </summary>
public Task ReadTask { get; }
/// <summary>
/// Connect to an existing remote end-point (server) and return the
/// <see cref="ConnectedEndPoint"/> object representing the new connection
/// </summary>
/// <param name="remoteEndPoint">The address of the remote end-point to connect to</param>
/// <param name="readCallback">The callback which will be called when a line of text is read from the newly-created connection</param>
/// <returns></returns>
public static ConnectedEndPoint Connect(IPEndPoint remoteEndPoint, Action<ConnectedEndPoint, string> readCallback)
{
Socket socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
socket.Connect(remoteEndPoint);
return new ConnectedEndPoint(socket, readCallback);
}
/// <summary>
/// Asynchronously accept a new connection from a remote end-point
/// </summary>
/// <param name="listener">The listening <see cref="Socket"/> which will accept the connection</param>
/// <param name="readCallback">The callback which will be called when a line of text is read from the newly-created connection</param>
/// <returns></returns>
public static async Task<ConnectedEndPoint> AcceptAsync(Socket listener, Action<ConnectedEndPoint, string> readCallback)
{
Socket clientSocket = await Task.Factory.FromAsync(listener.BeginAccept, listener.EndAccept, null);
return new ConnectedEndPoint(clientSocket, readCallback);
}
/// <summary>
/// Write a line of text to the connection, sending it to the remote end-point
/// </summary>
/// <param name="text">The line of text to write</param>
public void WriteLine(string text)
{
lock (_lock)
{
if (!_closing)
{
_writer.WriteLine(text);
_writer.Flush();
}
}
}
/// <summary>
/// Initiates a graceful closure of the connection
/// </summary>
public void Shutdown()
{
_Shutdown(SocketShutdown.Send);
}
/// <summary>
/// Implements <see cref="IDisposable.Dispose"/>
/// </summary>
public void Dispose()
{
_reader.Dispose();
_writer.Dispose();
_socket.Close();
}
/// <summary>
/// Constructor. Private -- use one of the factory methods to create new connections.
/// </summary>
/// <param name="socket">The <see cref="Socket"/> for the new connection</param>
/// <param name="readCallback">The callback for reading lines on the new connection</param>
private ConnectedEndPoint(Socket socket, Action<ConnectedEndPoint, string> readCallback)
{
_socket = socket;
Stream stream = new NetworkStream(_socket);
_reader = new StreamReader(stream, Encoding.UTF8, false, 1024, true);
_writer = new StreamWriter(stream, Encoding.UTF8, 1024, true);
ReadTask = _ConsumeSocketAsync(readCallback);
}
private void _Shutdown(SocketShutdown reason)
{
lock (_lock)
{
if (!_closing)
{
_socket.Shutdown(reason);
_closing = true;
}
}
}
private async Task _ConsumeSocketAsync(Action<ConnectedEndPoint, string> callback)
{
string line;
while ((line = await _reader.ReadLineAsync()) != null)
{
callback(this, line);
}
_Shutdown(SocketShutdown.Both);
}
}
A client program would use that class directly. The server side is encapsulated in another class, found in the same DLL with the above:
/// <summary>
/// Event arguments for the <see cref="ChatServer.Status"/> event
/// </summary>
public class StatusEventArgs : EventArgs
{
/// <summary>
/// Gets the status text
/// </summary>
public string StatusText { get; }
/// <summary>
/// Constructor
/// </summary>
/// <param name="statusText">The status text</param>
public StatusEventArgs(string statusText)
{
StatusText = statusText;
}
}
/// <summary>
/// A server implementing a simple line-based chat server
/// </summary>
public class ChatServer
{
private readonly object _lock = new object();
private readonly Socket _listener;
private readonly List<ConnectedEndPoint> _clients = new List<ConnectedEndPoint>();
private bool _closing;
/// <summary>
/// Gets a task representing the listening state of the servdere
/// </summary>
public Task ListenTask { get; }
/// <summary>
/// Raised when the server has status to report
/// </summary>
public event EventHandler<StatusEventArgs> Status;
/// <summary>
/// Constructor
/// </summary>
/// <param name="port">The port number the server should listen on</param>
public ChatServer(int port)
{
_listener = new Socket(SocketType.Stream, ProtocolType.Tcp);
_listener.Bind(new IPEndPoint(IPAddress.Any, port));
_listener.Listen(int.MaxValue);
ListenTask = _ListenAsync();
}
/// <summary>
/// Initiates a shutdown of the chat server.
/// </summary>
/// <remarks>This method closes the listening socket, which will subsequently
/// cause the listening task to inform any connected clients that the server
/// is shutting down, and to wait for the connected clients to finish a graceful
/// closure of their connections.
/// </remarks>
public void Shutdown()
{
_listener.Close();
}
private async Task _ListenAsync()
{
try
{
while (true)
{
ConnectedEndPoint client = await ConnectedEndPoint.AcceptAsync(_listener, _ClientReadLine);
_AddClient(client);
_CleanupClientAsync(client);
}
}
catch (ObjectDisposedException)
{
_OnStatus("Server's listening socket closed");
}
catch (IOException e)
{
_OnStatus($"Listening socket IOException: {e.Message}");
}
await _CleanupServerAsync();
}
private async Task _CleanupServerAsync()
{
ConnectedEndPoint[] clients;
lock (_lock)
{
_closing = true;
clients = _clients.ToArray();
}
foreach (ConnectedEndPoint client in clients)
{
try
{
client.WriteLine("Chat server is shutting down");
}
catch (IOException e)
{
_OnClientException(client, e.Message);
}
client.Shutdown();
}
// Clients are expected to participate in graceful closure. If they do,
// this will complete when all clients have acknowledged the shutdown.
// In a real-world program, may be a good idea to include a timeout in
// case of network issues or misbehaving/crashed clients. Implementing
// the timeout is beyond the scope of this proof-of-concept demo code.
try
{
await Task.WhenAll(clients.Select(c => c.ReadTask));
}
catch (AggregateException)
{
// Actual exception for each client will have already
// been reported by _CleanupClientAsync()
}
}
// Top-level "clean-up" method, which will observe and report all exceptions
// In real-world code, would probably want to simply log any unexpected exceptions
// to a log file and then exit the process. Here, we just exit after reporting
// exception info to caller. In either case, there's no need to observe a Task from
// this method, and async void simplifies the call (no need to receive and then ignore
// the Task object just to keep the compiler quiet).
private async void _CleanupClientAsync(ConnectedEndPoint client)
{
try
{
await client.ReadTask;
}
catch (IOException e)
{
_OnClientException(client, e.Message);
}
catch (Exception e)
{
// Unexpected exceptions are programmer-error. They could be anything, and leave
// the program in an unknown, possibly corrupt state. The only reasonable disposition
// is to log, then exit.
//
// Full stack-trace, because who knows what this exception was. Will need the
// stack-trace to do any diagnostic work.
_OnStatus($"Unexpected client connection exception. {e}");
Environment.Exit(1);
}
finally
{
_RemoveClient(client);
client.Dispose();
}
}
private void _ClientReadLine(ConnectedEndPoint readClient, string text)
{
_OnStatus($"Client {readClient.RemoteEndPoint}: \"{text}\"");
lock (_lock)
{
if (_closing)
{
return;
}
text = $"{readClient.RemoteEndPoint}: {text}";
foreach (ConnectedEndPoint client in _clients.Where(c => c != readClient))
{
try
{
client.WriteLine(text);
}
catch (IOException e)
{
_OnClientException(client, e.Message);
}
}
}
}
private void _AddClient(ConnectedEndPoint client)
{
lock (_lock)
{
_clients.Add(client);
_OnStatus($"added client {client.RemoteEndPoint} -- {_clients.Count} clients connected");
}
}
private void _RemoveClient(ConnectedEndPoint client)
{
lock (_lock)
{
_clients.Remove(client);
_OnStatus($"removed client {client.RemoteEndPoint} -- {_clients.Count} clients connected");
}
}
private void _OnStatus(string statusText)
{
Status?.Invoke(this, new StatusEventArgs(statusText));
}
private void _OnClientException(ConnectedEndPoint client, string message)
{
_OnStatus($"Client {client.RemoteEndPoint} IOException: {message}");
}
}
And that is, for the most part, all you need. The DLL code above is referenced (in my example) by two different programs, a server and a client.
Here's the server:
class Program
{
private const int _kportNumber = 5678;
static void Main(string[] args)
{
ChatServer server = new ChatServer(_kportNumber);
server.Status += (s, e) => WriteLine(e.StatusText);
Task serverTask = _WaitForServer(server);
WriteLine("Press return to shutdown server...");
ReadLine();
server.Shutdown();
serverTask.Wait();
}
private static async Task _WaitForServer(ChatServer server)
{
try
{
await server.ListenTask;
}
catch (Exception e)
{
WriteLine($"Server exception: {e}");
}
}
}
And here's the client:
class Program
{
private const int _kportNumber = 5678;
static void Main(string[] args)
{
IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Loopback, _kportNumber);
ConnectedEndPoint server = ConnectedEndPoint.Connect(remoteEndPoint, (c, s) => WriteLine(s));
_StartUserInput(server);
_SafeWaitOnServerRead(server).Wait();
}
private static void _StartUserInput(ConnectedEndPoint server)
{
// Get user input in a new thread, so main thread can handle waiting
// on connection.
new Thread(() =>
{
try
{
string line;
while ((line = ReadLine()) != "")
{
server.WriteLine(line);
}
server.Shutdown();
}
catch (IOException e)
{
WriteLine($"Server {server.RemoteEndPoint} IOException: {e.Message}");
}
catch (Exception e)
{
WriteLine($"Unexpected server exception: {e}");
Environment.Exit(1);
}
})
{
// Setting IsBackground means this thread won't keep the
// process alive. So, if the connection is closed by the server,
// the main thread can exit and the process as a whole will still
// be able to exit.
IsBackground = true
}.Start();
}
private static async Task _SafeWaitOnServerRead(ConnectedEndPoint server)
{
try
{
await server.ReadTask;
}
catch (IOException e)
{
WriteLine($"Server {server.RemoteEndPoint} IOException: {e.Message}");
}
catch (Exception e)
{
// Should never happen. It's a bug in this code if it does.
WriteLine($"Unexpected server exception: {e}");
}
}
}
In my opinion, one of the most important things for you to note in the above is that the ConnectedEndPoint and ChatServer classes have zero dependency on the classes which use them. Through the use of callback delegates and events, the code that depends on these classes are able to interact bi-directionally without these supporting classes having to know about the types that code resides in (see "inversion of control", which this is a variation of).
The more you can make your code relationships look like a tree with only single-direction references, the easier it will be to write the code, and to maintain it later.
Note: I used both events and callback delegates for the sake of illustration. Either approach works fine on its own. The main trade-offs are complexity vs. flexibility. Using events makes the code more flexible — event handlers can be added and removed as necessary — but if one implements events using the .NET convention of a method signature with a sender and an EventArgs parameter, it is somewhat more "heavy-weight" than just passing a simple callback delegate when creating the object in question. I put an example of each in the code, and you can decide which approaches you prefer in which situations.
You'll also note that the above makes heavy use of C#'s asynchronous features. At first, this may make the code seem harder to read. But in fact, it's actually much easier to get everything working using these features, than if I were to try to use the older BeginXXX()/EndXXX() methods or, heaven forbid, dedicate a single thread to each connection (which scales very poorly as the client count goes up). It's absolutely worth getting used to thinking of operations which are inherently asynchronous, such as network I/O, in this way.
| 归档时间： |
|
| 查看次数： |
2613 次 |
| 最近记录： |