如何用Ruby CZTop::Dealer实现异步客户端模式与心跳检测？

ZeroMQ 的 DEALER 套接字在请求响应模式中常被当作异步客户端使用。与 REQ 不同，DEALER 不维护发送与接收的严格顺序，客户端可以连续发出多个请求，再根据返回帧中的标识进行匹配。这种能力让它在需要并发、低延迟和自定义心跳的场景中非常实用。Ruby 通过 CZTop 这个 gem 提供了对 DEALER 套接字的封装，开发者可以直接用面向对象的方式完成连接、发送、接收和轮询操作。

一、DEALER 与 REQ 的核心差异

REQ 套接字在 ZeroMQ 中是一个同步请求方，它强制客户端必须按照发送请求、等待响应、再发送下一个请求的顺序工作。如果服务端处理较慢，客户端只能阻塞等待，这在异步架构中非常受限。DEALER 则去掉了这层状态限制，它将消息发送之后立即返回，不等待服务端应答。多个请求可以同时进入网络，服务端的响应可以以任意顺序返回，客户端通过消息帧中的地址信息或业务 ID 进行区分。

这种无状态设计让 DEALER 非常适合作为异步客户端的基础。例如在一个网关服务中，需要同时向多个后端实例发送查询，每个实例的响应时间不同，使用 REQ 只能串行等待，而 DEALER 可以先把所有请求都发出去，再统一接收。实现时要注意，正因为 DEALER 不再自动维护请求与响应的一一对应关系，所以应用层必须自己设计消息标识，否则会出现响应串位的现象。

在 ZeroMQ 体系中，DEALER 通常与 ROUTER 配合使用。ROUTER 会为每一个连接的 DEALER 生成或读取一个身份标识，转发消息时会在首帧添加该标识。客户端需要使用 options.identity 设置稳定身份，这样服务端才能正确回复，心跳检测也才能知道消息来自哪个连接。

二、初始化 CZTop::Dealer 并完成基础收发

使用 CZTop 之前需要安装 gem，并在 Ruby 文件中引入。CZTop 的 API 设计比较直接，CZTop::Socket::DEALER.new 可以创建一个 DEALER 套接字，接着通过 connect 方法连接到服务端的端点。如果客户端运行在网关后面，还需要通过 options.identity 指定一个稳定的字符串身份，避免服务端因为临时身份变化而无法识别。

下面是一个最基础的连接和收发示例。这里使用阻塞的 receive 方法仅用于演示消息帧结构，实际异步客户端不应在单线程中长时间阻塞。

require "cztop"

dealer = CZTop::Socket::DEALER.new
dealer.options.identity = "async-client-1"
dealer.connect("tcp://127.0.0.1:5555")

# 发送一条心跳消息
dealer.send(CZTop::Message.new("ping"))

# 阻塞等待响应，仅用于演示
reply = dealer.receive
puts reply[0].to_s

代码中的 CZTop::Message 表示一个多帧消息。DEALER 发送时可以将多个字符串按顺序组装成一帧一帧的数据，接收时会得到一个可以按索引访问的 CZTop::Message 对象。这里 reply[0] 取第一帧，通常就是服务端返回的文本内容。需要注意的是，真实业务中不要直接在事件循环里调用阻塞 receive，应该使用轮询器让线程同时处理心跳、业务响应和超时判断。

连接端点的格式使用 ZeroMQ 标准的 tcp://、ipc:// 或 inproc://。异步客户端最常见的是 TCP 连接，地址为 tcp://127.0.0.1:5555 这种形式。如果服务端位于远程主机，把 127.0.0.1 替换为真实 IP 或域名即可，端口需要与服务端监听保持一致。

三、ping/pong 心跳检测的实现

心跳检测是异步客户端中不可缺少的机制。DEALER 本身不知道连接是否已经断开，尤其是对端进程崩溃或网络链路中断时，TCP 连接可能不会立刻报告错误。如果客户端一直不发送消息，可能无法及时发现服务端已不可用。因此需要在应用层周期性地发送轻量级的 ping 消息，并等待服务端返回 pong。心跳间隔和超时时间要根据实际网络延迟调整，通常间隔 3 到 10 秒，超时设置为间隔的 2 到 3 倍。

下面的类把心跳发送、pong 处理和时间判断整合在一起。每次收到 pong 后更新 @last_pong 时间戳，主循环根据该时间戳决定是否继续发送 ping 或判定超时。

require "cztop"

class DealerHeartbeatClient
  HEARTBEAT_INTERVAL = 5.0
  HEARTBEAT_TIMEOUT = 2.0

  def initialize(endpoint, identity: "client-1")
    @endpoint = endpoint
    @identity = identity
    @socket = CZTop::Socket::DEALER.new
    @socket.options.identity = @identity
    @socket.connect(@endpoint)
    @poller = CZTop::Poller.new(@socket)
    @last_pong = Time.now.to_f
  end

  def run
    loop do
      wait_for_events
      send_heartbeat_if_needed
      check_heartbeat_timeout
    end
  end

  private

  def wait_for_events
    readable = @poller.wait(1000)
    if readable.include?(@socket)
      message = @socket.receive
      process_frames(message)
    end
  end

  def process_frames(message)
    if message && message[0].to_s == "pong"
      @last_pong = Time.now.to_f
      puts "pong received"
    else
      handle_business_message(message)
    end
  end

  def send_heartbeat_if_needed
    if Time.now.to_f - @last_pong > HEARTBEAT_INTERVAL
      @socket.send(CZTop::Message.new("ping"))
    end
  end

  def check_heartbeat_timeout
    if Time.now.to_f - @last_pong > HEARTBEAT_TIMEOUT * 3
      puts "server lost, reconnecting"
      reconnect
    end
  end

  def reconnect
    @socket.close
    @socket = CZTop::Socket::DEALER.new
    @socket.options.identity = @identity
    @socket.connect(@endpoint)
    @poller = CZTop::Poller.new(@socket)
    @last_pong = Time.now.to_f
  end

  def handle_business_message(message)
    # 根据业务协议解析多帧消息
    if message
      puts "business message: #{message[0].to_s}"
    end
  end
end

代码中的 @poller.wait(1000) 表示最多等待 1000 毫秒，如果在此期间套接字没有可读数据就返回，这样循环就能继续执行心跳发送和超时检查。这种“时间片轮询”的方式比单独开一个心跳线程更简单，也避免了多线程间共享套接字带来的并发问题。在单线程模型下，所有消息按顺序处理，业务消息和心跳消息不会互相打断。

服务端需要在 ROUTER 或 REP 等套接字上监听 ping 并返回 pong。一种常见做法是服务端在收到首帧为 ping 的消息时，直接发送一个只包含 pong 的应答。这样客户端就能通过 message[0].to_s == "pong" 快速识别。更复杂的协议可以将 pong 放在第二帧，第一帧保存客户端身份，具体根据服务端实现调整。

四、轮询、超时与重连的注意事项

使用轮询器时，一个容易忽略的问题是 CZTop::Poller 与原生 ZeroMQ 轮询器一样，只返回可读或可写的套接字，不会主动抛出异常。因此在 wait 返回后可读集合为空时，循环仍然需要推进心跳逻辑。上面示例中的 wait_for_events 会在没有事件时直接返回，然后继续执行 send_heartbeat_if_needed 和 check_heartbeat_timeout，从而保持心跳的节奏。

超时判定不能只看单次心跳的响应时间。比如网络抖动导致某次 ping 的 pong 晚到了 500 毫秒，但服务端实际仍然健康。如果超时时间设置得过短，就会误判并频繁重连。一般建议设置超时为心跳间隔的 2 倍以上，并且允许连续 2 到 3 次未收到 pong 再触发重连。可以用一个计数器记录连续失败的次数，收到一次 pong 就清零，这样比简单比较时间戳更稳健。

重连操作需要先关闭旧套接字，再重新创建、设置身份并连接。注意不要在 reconnect 中直接复用旧的 @socket，因为 ZeroMQ 的套接字与连接状态绑定，简单的 connect 不会清除已经断开的底层连接。重建套接字虽然有一定开销，但对于异步客户端来说通常可以接受。如果重连成功后服务端还保留旧身份的会话状态，可以通过重新登录或注册来恢复业务上下文。

业务消息在重连窗口内可能会发送失败。为了减少消息丢失，可以在客户端维护一个待确认队列，当消息发送后未收到应用层 ACK 或响应时，先将该消息保留在内存中，待连接恢复后重新发送。这个队列与心跳机制独立，但共用同一个轮询循环，因此不会引入更多线程复杂度。

Ruby CZTopDealer套接字心跳检测修改时间：2026-09-18 07:56:37