我对 Debian Wheezy 上的 Mono 4.8 上的 NetMQ 4.0.0.1 有一些问题。
在我不会停止调用它发送新消息之前，经销商套接字不会发送任何消息。当我将Thread.Sleep( 1000 )在创建任务之间放置时，一切都很好。我想承认一切都在 .Net Framework 4.5 和 .Net Core 1.1 上的 Windows 上运行，没有任何Thread.Sleep().
我有这样的模式：
我已经添加了调试消息，我可以看到我正在REQ循环中的任务中创建 100 个套接字，并且路由器正在队列中获取请求，而不是通过经销商发送它们，并且 TCP 的另一端没有发生任何事情，直到我会在 REQ 套接字上停止调用发送。每 5 个任务中的一个简单Thread.Sleep()的工作。它看起来像 Poller 错误，或 Dealer 错误，或者我做错了什么。
这是中间框的代码：
public class CollectorDevice : IDisposable
{
private NetMQPoller _poller;
private RouterSocket _frontendSocket;
private DealerSocket _backendSocket;
private readonly string _backEndAddress;
private readonly string _frontEndAddress;
private readonly int _expectedFrameCount;
private readonly ManualResetEvent _startSemaphore = new ManualResetEvent(false);
private readonly Thread _localThread;
private static Logger _logger = LogManager.GetCurrentClassLogger();
/// <summary>
/// Constructor
/// </summary>
/// <param name="backEndAddress"></param>
/// <param name="frontEndAddress"></param>
/// <param name="expectedFrameCount"></param>
public CollectorDevice(string backEndAddress, string frontEndAddress, int expectedFrameCount)
{
_expectedFrameCount = expectedFrameCount;
_backEndAddress = backEndAddress;
_frontEndAddress = frontEndAddress;
_localThread = new Thread(DoWork) { Name = "IPC Collector Device Thread" };
}
public void Start()
{
_localThread.Start();
_startSemaphore.WaitOne();
}
public void Stop()
{
_poller.Stop();
}
#region Implementation of IDisposable
public void Dispose()
{
Stop();
}
#endregion
#region Private Methods
private void DoWork()
{
try
{
using (_poller = new NetMQPoller())
using (_frontendSocket = new RouterSocket(_frontEndAddress))
using (_backendSocket = new DealerSocket(_backEndAddress))
{
_backendSocket.ReceiveReady += OnBackEndReady;
_frontendSocket.ReceiveReady += OnFrontEndReady;
_poller.Add(_frontendSocket);
_poller.Add(_backendSocket);
_startSemaphore.Set();
_poller.Run();
}
}
catch (Exception e)
{
_logger.Error(e);
}
}
private void OnBackEndReady(object sender, NetMQSocketEventArgs e)
{
NetMQMessage message = _backendSocket.ReceiveMultipartMessage(_expectedFrameCount);
_frontendSocket.SendMultipartMessage(message);
}
private void OnFrontEndReady(object sender, NetMQSocketEventArgs e)
{
NetMQMessage message = _frontendSocket.ReceiveMultipartMessage(_expectedFrameCount);
_backendSocket.SendMultipartMessage(message);
}
#endregion
}
这是客户端：
class Program
{
private static Logger _logger = LogManager.GetCurrentClassLogger();
private static void Main(string[] args)
{
Console.WriteLine("Client. Please enter message for server. Enter 'QUIT' to turn off server");
Console.ReadKey();
using (var collectorDevice = new CollectorDevice(">tcp://localhost:5556", "inproc://broker", 3))
{
collectorDevice.Start();
List<Task> tasks = new List<Task>();
for (int i = 0; i < 100; i++)
{
Console.WriteLine(i);
int j = i;
Task t = Task.Factory.StartNew(() =>
{
try
{
using (var req = new RequestSocket("inproc://broker"))
{
req.SendFrame(String.Format("Request client: {0} id: {1}", j, Task.CurrentId));
_logger.Debug(String.Format("Request client: {0} id: {1}", j, Task.CurrentId));
Console.WriteLine(String.Format("Request client: {0} id: {1}", j, Task.CurrentId));
string responseMessage = req.ReceiveFrameString();
_logger.Debug(String.Format("Response from server: {0} id: {1} message: {2}", j, Task.CurrentId, responseMessage));
Console.WriteLine(String.Format("Response from server: {0} id: {1} message: {2}", j, Task.CurrentId, responseMessage));
}
}
catch (Exception e)
{
Console.WriteLine(e);
_logger.Error(e);
}
});
tasks.Add(t);
//Thread.Sleep (100);//<- This thread sleep is fixing problem?
}
Task.WaitAll(tasks.ToArray());
}
}
}
和服务器端：
class Program
{
private static Logger _logger = LogManager.GetCurrentClassLogger();
static void Main(string[] args)
{
try{
using (var routerSocket = new RouterSocket("@tcp://*:5556"))
{
var poller = new NetMQPoller();
routerSocket.ReceiveReady += RouterSocketOnReceiveReady;
poller.Add(routerSocket);
poller.Run();
}
}
catch(Exception e)
{
Console.WriteLine (e);
}
Console.ReadKey ();
}
private static void RouterSocketOnReceiveReady(object sender, NetMQSocketEventArgs netMqSocketEventArgs)
{
NetMQMessage clientMessage = new NetMQMessage();
bool result = netMqSocketEventArgs.Socket.TryReceiveMultipartMessage(new TimeSpan(0, 0, 0, 5),
ref clientMessage, 5);
if (result == false)
{
Console.WriteLine ("Something went wrong?!");
}
var address = clientMessage[0];
var address2 = clientMessage[1];
var clientMessageString = clientMessage[3].ConvertToString();
//_logger.Debug("Message from client received: '{0}'", clientMessageString);
Console.WriteLine (String.Format ("Message from client received: '{0}'", clientMessageString));
netMqSocketEventArgs
.Socket.SendMoreFrame(address.Buffer)
.SendMoreFrame(address2.Buffer)
.SendMoreFrameEmpty()
.SendFrame("I have received your message");
}
}
有人有什么想法吗？
我在想我可能正在使用来自不同线程的套接字，所以我将它打包到 ThreadLocal 结构中，但它没有帮助。比我读到一些与 NetMQ 统一的问题，所以我添加了'AsyncIO.ForceDotNet.Force() ;' 在每个套接字构造函数调用之前，这也没有帮助。比我将我的单声道从 4.4 更新到 4.8 并且它看起来仍然一样。
我检查了任务之间的 Thread.Sleep(100) 是否正在解决问题。它很奇怪