Rails集成Kafka Python消费者/生产者阻塞问题求助

我正在开发一个Rails服务器，需要运行作为Kafka消费者/生产者的Python脚本。服务器要执行脚本并从消费者接收处理后的数据，渲染到网站上。目前能运行脚本，但解决不了消费者连接的问题：消费者要么持续监听消息，要么在while循环里运行，从Ruby启动消费者后会一直阻塞，没法执行其他脚本。

理想的消息流是：已登录用户的邮箱→Kafka生产者→消息队列→Kafka消费者生成数据写入数据库→查询生产者从数据库取数→消息队列→消费者接收数据并渲染到网站。

我试过用一个管理脚本完成所有工作，但因为消费者持续监听导致脚本无法正常运行。以下是相关代码：

生产者1代码 #

from kafka import KafkaProducer
from faker import Faker
import json
import time
class producer1():

        '''
        fr_instance= Faker()
        def get_new_user():
        return {"email_address":fr_instance.email(),"first_name": fr_instance.first_name(),
        "lastname": fr_instance.last_name(), "occupation":fr_instance.job()}
        '''
        def __init__(self):
                self
                

        def json_serializer(self, data):
                return json.dumps(data).encode("utf-8")



        def send(self,email):
                print(email)
                producer = KafkaProducer(bootstrap_servers='localhost:9092',
                        value_serializer = self.json_serializer)
                registred_user = {"email": email}
                future = producer.send("NewUserTopic", registred_user)
                print (registred_user)
                result = future.get(timeout=10)
                

p = producer1()
if __name__ == '__main__':   
        email = "testmail@aaaaaaaa.com"
        p.send(email)

第一个消费者代码 #

from kafka import KafkaConsumer
import json
import random
from sqlalchemy.orm import sessionmaker
import dbservice
import time

class consumer1():
    def __init__(self) -> None:
        self
    def email(self):
        consumer = KafkaConsumer('NewUserTopic',
        bootstrap_servers='localhost:9092',
        auto_offset_reset = 'latest', enable_auto_commit= False)
        for msg in consumer:
            
            
            msg_out = json.loads(msg.value)
            for value in  msg_out.values():
                

            
            #return print(msg_out)
                return (value)
            
    #generate dummy address , eth
    def gen_ETHw (self):
         numbers = str(random.randint(11111,99999))
         wallet_num = str("Ox"+numbers)
         return (wallet_num)
    #generate dummy address , btc
    def gen_BTCw (self):

         numbers = str(random.randint(11111,99999))
         wallet_num = str("Ox"+numbers)
         return (wallet_num)
    
    def commit_db (self, email, ETHw, BTCw):
        Session = sessionmaker(bind=dbservice.engine)
        s = Session()
        
        input = dbservice.walletdb( email,ETHw, BTCw)
        time.sleep(2)
        s.add(input)
        s.commit()
    

         
if __name__ =='__main__':
    while True:
        c = consumer1()
        c.commit_db(c.email(),c.gen_ETHw(),c.gen_BTCw())

查询生产者代码 #

import dbservice
import dbservice
from sqlalchemy.orm import sessionmaker
from kafka import KafkaProducer
import json


class query_prod ():
        def __init__(self, email) -> None:
                self = self
                self.email = email
               
        
        def json_serializer(data):
                return json.dumps(data).encode("utf-8")

        producer = KafkaProducer(bootstrap_servers='localhost:9092',
                        value_serializer = json_serializer)

        Session = sessionmaker(bind=dbservice.engine)
        s = Session()
        
        def query_address(self,email):
                Session = sessionmaker(bind=dbservice.engine)
                s = Session()
                for s in s.query(dbservice.walletdb).filter_by(email=email):
                       return {"email":s.email,"ETH_w":s.ETH_w,"BTC_w":s.BTC_w}

        def send(self, email):
                data_to_send = self.query_address(email)
        
                future = self.producer.send("QueryAdressToServer", data_to_send)
                print (data_to_send)
                result = future.get(timeout=10)

if __name__ == '__main__':
        email = "testmail@aaaaaaaa.com"
        query_prod=query_prod(email)
        query_prod.send(email)

网站数据接收消费者代码 #

from kafka import KafkaConsumer
import json
import time

class consume_for_web():
    string=""
    def __init__(self) -> None:
        self = self
        string = self.string
    
    
    def consumer(self):
        consumer = KafkaConsumer('QueryAdressToServer',
            bootstrap_servers='localhost:9092',
            auto_offset_reset = 'latest', enable_auto_commit= False)
        print('starting consumer')
        for msg in consumer:
            data = (('{}'.format(json.loads(msg.value))))
            self.string = self.string + data
            return print(data)
        
    def read_str(self):
        return print(self.string)

if __name__ =='__main__':
    while True:
        c = consume_for_web()
        c.consumer()
        ##print("reading")
        #c.read_str()

Rails Pages控制器代码 #

class PagesController < ApplicationController
  def home
  end

  def about
  end

  before_action :require_login
  def genw
    our_input = current_user.email
    puts our_input
    
    @consumer_result = `python3 /Users/samuelrybar/python_projects/Kafka_demo1/kafka-prod-coms/consumer2.py`

  end

  
  def mywa
  end
  def save
    
  end
end

1. 让消费者脱离Rails请求周期运行 #

Rails请求是同步阻塞的，绝对不能在请求流程里运行持续监听的Kafka消费者。应该把消费者作为独立后台进程运行：

• 用nohup python3 consumer1.py &让脚本在后台常驻
• 配置成systemd服务，确保进程崩溃后自动重启

2. 优化消费者逻辑，避免资源浪费 #

第一个消费者的while True会不断创建新实例，完全没必要。修改为单例消费者持续监听：

if __name__ =='__main__':
    c = consumer1()
    consumer = KafkaConsumer('NewUserTopic',
        bootstrap_servers='localhost:9092',
        auto_offset_reset='latest',
        enable_auto_commit=False)
    for msg in consumer:
        msg_out = json.loads(msg.value)
        email = next(iter(msg_out.values()))
        eth_w = c.gen_ETHw()
        btc_w = c.gen_BTCw()
        c.commit_db(email, eth_w, btc_w)
        consumer.commit()  # 手动提交偏移量，避免重复消费

3. Rails直接查询数据库，简化流程 #

你的消费者已经把数据写入数据库，Rails完全可以直接查询数据库获取结果，不需要再通过Kafka绕圈。修改genw方法：

def genw
  our_input = current_user.email
  # 假设对应ActiveRecord模型为Wallet
  @wallet = Wallet.find_by(email: our_input)
  # 触发生产者发送消息时，用&让脚本后台运行，不阻塞请求
  system("python3 /path/to/producer1.py #{our_input} &")
end

4. 用异步任务处理Kafka交互 #

如果必须通过Kafka传递数据，可借助Sidekiq等异步框架：

• 创建Sidekiq任务负责调用Python生产者发送消息
• 消费者写入数据库后，Rails页面通过ActionCable（WebSocket）实时获取更新，或者前端轮询接口

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