NETMQ Pub/Sub订阅者首次运行正常，重连后无法接收消息求助

a. 先启动订阅端是否正常？ #

这是NetMQ基于ZeroMQ原生Pub/Sub模式的默认正常行为。普通PUB套接字绑定后，只会将消息推送给在它开始发送消息前就完成连接订阅的客户端。如果发布端先启动，后续连接的订阅端无法被发布端的订阅列表收录，自然收不到消息。

b. 实现订阅端随时重连接收消息的方案 #

方案1：使用XPublisherSocket替代普通PUB套接字 #

XPUB是扩展发布套接字，它能主动感知新连接的订阅者，并维护动态的订阅列表。即使发布端先启动，后续重启或新启动的订阅端连接后，也能正常接收后续的消息。

发布端代码示例：

using NetMQ;
using NetMQ.Sockets;

// 发布端使用XPublisherSocket
using (var pubSocket = new XPublisherSocket())
{
    pubSocket.Bind("tcp://*:5555"); // 绑定端口
    
    while (true)
    {
        // 模拟发送机器学习采集的数据
        var data = "ML data sample: " + DateTime.Now.Ticks;
        pubSocket.SendFrame(data);
        Thread.Sleep(500);
    }
}

方案2：给订阅端配置自动重连参数 #

订阅端使用SubscriberSocket时，通过设置重连间隔参数，让它在断开后自动尝试重连。配合XPUB使用效果最佳，即使单独使用也能提升重连成功率。

订阅端代码示例：

using NetMQ;
using NetMQ.Sockets;

using (var subSocket = new SubscriberSocket())
{
    // 设置自动重连参数
    subSocket.Options.ReconnectInterval = TimeSpan.FromMilliseconds(100);
    subSocket.Options.ReconnectIntervalMax = TimeSpan.FromSeconds(5);
    
    subSocket.Connect("tcp://localhost:5555");
    subSocket.Subscribe(""); // 订阅所有消息
    
    Console.WriteLine("订阅端已启动，等待接收消息...");
    
    while (true)
    {
        if (subSocket.TryReceiveFrameString(out var message))
        {
            Console.WriteLine("收到消息: " + message);
        }
    }
}

额外注意事项 #

• 不要在发布端依赖Thread.Sleep来"等待订阅端连接"，这种方式不可靠，无法应对订阅端重启的场景。
• 如果需要订阅端能接收重启前的历史消息，还需要额外实现消息缓存机制，比如在发布端暂存最近的N条消息，当新订阅者连接时主动补发。

内容的提问来源于stack exchange，提问作者Sixk