这是我在这里的第一个问题，我的名字是安娜！
我的问题： 我的客户有几个个人设备（带 gps 的黑匣子）来定位人/汽车...大约有 12000 人/汽车使用该设备...它将他们的位置发送到指定的 IP/端口...我无能为力那边……
我的工作？ 开发一个侦听器以捕获设备发送的所有数据并使用 .NET 将其加载到数据库中...
我的想法： 使用线程的窗口服务（可能是 ThreadPool？）。所以服务会捕获所有传入的消息，创建一个线程并放入数据库中......
这是解决问题的最佳方法吗？ 我在这里阅读有关消息队列 (MSMQ) 的信息...您认为我应该使用它吗？
安娜
【问题讨论】：
标签： c# .net asp.net multithreading msmq
以下代码未经测试或仅应被视为指导。 它使用线程池来处理所有事情。我将所有内容都放在同一个文件中，仅用于示例。您应该将所有内容分成几个文件。
重要的是不要让所有客户端直接保存到数据库，因为这会使线程池饿死。试验“MaxThreads”常量以获得适用于您的数据库/服务器的值。
还要记住，我在下面根本不处理任何异常。例如，您需要在 BeginRead、EndRead 和 TcpListener 方法上处理 SocketException。
我尝试使用最少数量的线程同步锁，代码应该非常高效。瓶颈很可能是数据库。
using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Threading;
namespace FastListener
{
/// <summary>
/// Example position class, replace with a real definition
/// </summary>
public class Position
{
public int X { get; set; }
public int Y { get; set; }
}
/// <summary>
/// Needed to be able to pass socket/buffer info
/// between asynchronous requests.
/// </summary>
public struct ClientContext
{
public Socket socket;
public byte[] buffer;
}
class Program
{
/// <summary>
/// Positions received from mobile clients but not yet saved
/// into the database.
/// </summary>
private readonly Queue<Position> _positions = new Queue<Position>();
/// <summary>
/// Number of threads currently saving stuff to the database.
/// </summary>
private int _poolThreads;
/// <summary>
/// Maximum number of threads that can save info to the database.
/// </summary>
private const int MaxThreads = 10;
static void Main(string[] args)
{
new Program().Start();
}
private void Start()
{
TcpListener listener = new TcpListener(IPAddress.Any, 1343);
listener.Start(50);
listener.BeginAcceptSocket(OnAccept, listener);
}
// Listener got a new connection
private void OnAccept(IAsyncResult ar)
{
TcpListener listener = (TcpListener) ar.AsyncState;
// It's very important to start listening ASAP
// since you'll have a lot of incoming connections.
listener.BeginAcceptSocket(OnAccept, listener);
// I recommend that you create a buffer pool to improve performance
byte[] buffer = new byte[400];
// Now accept the socket.
Socket socket = listener.EndAcceptSocket(ar);
StartRead(new ClientContext {buffer = buffer, socket = socket});
}
private void StartRead(ClientContext context)
{
// start reading from the client.
context.socket.BeginReceive(context.buffer, 0, 400, SocketFlags.None, OnReceive, context);
}
// Stuff from a client.
private void OnReceive(IAsyncResult ar)
{
ClientContext context = (ClientContext) ar.AsyncState;
int bytesRead = context.socket.EndReceive(ar);
if (bytesRead == 0)
{
// Put the buffer back in the pool
context.socket.Close();
return;
}
// convert bytes to position.
// i'll just fake that here.
Position pos = new Position();
// Either handle the request directly
if (_poolThreads < MaxThreads)
ThreadPool.QueueUserWorkItem(SaveToDatabase, pos);
else
{
// Or enqueue it to let a already active
// thread handle it when done with the previous position
lock (_positions)
_positions.Enqueue(pos);
}
// Don't forget to read from the client again
StartRead(context);
}
// will save stuff to the database.
private void SaveToDatabase(object state)
{
// Could use Interlocked.Increment, but not really vital if
// one more more extra threads are saving to the db.
++_poolThreads;
Position position = (Position) state;
while (true)
{
// IMPLEMENT DB SAVE LOGIC HERE.
// check if another position is in the queue.
lock (_positions)
{
if (_positions.Count > 0)
position = _positions.Dequeue();
else
break; // jump out of the loop
}
}
--_poolThreads;
}
}
}
【讨论】：
位置/时间的数量，以及位置信息的传输方式（是 TCP、UDP 等）将有助于确定最佳方法。
一些问题：
从它的声音来看，拥有一个可以捕获请求并只是在内部维护一个队列以保存到数据库的服务应该可以正常工作。我不相信 MSMQ 对你有用，除非你可以改变传输端，即使那样，它可能有也可能没有必要。
编辑：鉴于下面的 cmets，我建议您有一个 TCP 侦听器将请求传递到线程池以进行处理。
我会看一下this tutorial 关于使用线程池设置 TCP 侦听服务器的信息。我看到的最大潜在问题是请求的数量——从你所说的来看，你将有大约 400 个请求/秒。如果没有一个非常好的系统，这将是一个挑战。线程池可能会比尝试执行自己的线程执行得更好，因为您希望避免必须不断创建新线程的开销。您肯定希望在主处理循环中几乎没有延迟（例如 Sleep(0)，或者根本不睡眠），因为平均每 2.5 毫秒会有一个请求。单次睡眠往往会在最短 14-15 毫秒的时间内进行时间片，因此您可能不希望在循环中进行任何睡眠。
不过，我想说的是，您可能会发现这不太好用，因为原始连接数量可能存在问题。如果有任何方法可以转换为正在发送的 UDP 数据包，它可能会为您带来更好的吞吐量（以牺牲一些可靠性为代价）。
【讨论】：