如何用Ruby以同步模式（ack=1或ack=all）向Kafka生产数据

你不需要绕Redis做中转——rdkafka-ruby可以通过等待消息确认的方式实现业务层面的同步推送，这是你可能忽略的核心点。虽然它底层是异步生产模型，但通过阻塞等待Kafka的delivery report（消息确认），完全能达到"消息成功写入后再继续执行"的同步效果。

一、rdkafka-ruby 同步推送实现示例 #

首先安装依赖：

gem install rdkafka

以下是完整的同步发送代码：

require 'rdkafka'
require 'thread' # 引入线程相关类

# 初始化生产者配置
producer_config = Rdkafka::Config.new(
  "bootstrap.servers": "your-kafka-broker:9092",
  "acks": "all",          # 要求所有同步副本确认，确保最高可靠性
  "retries": 3,           # 发送失败自动重试3次
  "queue.buffering.max.ms": 0 # 禁用批量缓冲，消息立即发送
)

# 创建生产者实例
producer = producer_config.producer

# 封装同步发送方法
def sync_send(producer, topic, payload)
  # 用条件变量和互斥锁实现阻塞等待
  mutex = Mutex.new
  cond = ConditionVariable.new
  delivery_success = false
  error_msg = nil

  # 发送消息并绑定确认回调
  producer.produce(
    topic: topic,
    payload: payload,
    callback: lambda do |report|
      mutex.synchronize do
        if report.error
          error_msg = report.error.to_s
        else
          delivery_success = true
        end
        cond.signal # 通知主线程已收到确认
      end
    end
  )

  # 触发生产者处理发送队列与回调（必须调用，否则回调不会执行）
  producer.poll(0)

  # 阻塞等待确认，设置10秒超时（可根据业务调整）
  mutex.synchronize do
    unless cond.wait(mutex, 10)
      raise "消息发送超时（10秒）"
    end
    raise "消息发送失败: #{error_msg}" unless delivery_success
  end
end

# 实际调用示例
begin
  sync_send(producer, "test_topic", "这是一条同步推送的消息")
  puts "消息已成功写入Kafka"
rescue StandardError => e
  puts "发送失败: #{e.message}"
ensure
  # 关闭生产者，确保所有未完成的消息被处理
  producer.close
end

代码关键点说明： #

• acks: "all"：强制要求Kafka集群中所有同步副本都确认消息，避免丢消息
• queue.buffering.max.ms: 0：关闭批量发送缓冲，让消息立即被推送，消除异步延迟
• 利用ConditionVariable实现阻塞：主线程会等待直到收到Kafka的确认回调，或超时抛出异常
• producer.poll(0)：触发生产者内部的事件循环，处理发送队列和回调逻辑，这是rdkafka-ruby的必要操作

二、其他可选方案 #

如果对rdkafka-ruby的伪同步方案有顾虑，还可以考虑：

• ruby-kafka社区分支：虽然官方版本停更，但部分社区fork已经适配了Kafka 2.x及以上版本，可以自行搜索验证
• Kafka REST Proxy：通过HTTP POST请求同步推送消息，无需依赖Ruby Kafka库，但性能会略低于原生客户端

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