kafka 消费者到 ruby​​ on rails

Question

我目前正在使用一个应该运行 python 脚本的 rails 服务器，它是 kafka 消费者/生产者。

服务器必须运行脚本，然后从消费者那里接收处理后的数据，并将它们呈现给站点。

我能够运行脚本，但无法为连接消费者的解决方案提供资金。因为消费者要么不停地运行接受消息，要么在 while 循环中运行。我尝试首先从 ruby​​ 运行消费者，它启动消费者，但从未获得消费者，因为它正在侦听，但其他脚本无法运行。

所以理想的消息流应该是这样的 -> 从登录用户到 kafka 生产者的电子邮件 -> MQ -> kafka 消费者生成数据写入 db -> 生产者从数据库查询数据 -> MQ -> 消费者接受数据并将它们呈现到站点。

理想的场景是一个脚本，让我们称之为管理器，它完成所有工作，只接受数据并返回它。它也无法做到这一点，因为一个脚本也运行消费者并监听生产者，但它从未运行过。

所以这是我的代码：

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 querry_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 querry_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.querry_address(email)
        
                future = self.producer.send("QuerryAdressToServer", data_to_send)
                print (data_to_send)
                result = future.get(timeout=10)

if __name__ == '__main__':
        email = "testmail@aaaaaaaa.com"
        querry_prod=querry_prod(email)
        querry_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('QuerryAdressToServer',
            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

感谢您的时间和帮助，我真的很感激。 :))

Answer 1

不知道为什么要尝试从正在运行的 Rails 服务器运行 python 脚本。对我来说，这听起来是个非常糟糕的主意。您可以直接从 Ruby 和 Ruby on Rails 运行 Kafka 消费者/生产者。我建议你调查Karafka。我们一直在工作中成功使用它。