self.request_topic='requestRemoteModelServer'
self.response_topic='responseRemoteModelServer'
self.producer = Biz_模型控制服务.kafka_producer_init()
self.consumer=Biz_模型控制服务.kafka_consumer_init(self.response_topic,10000)
Biz_模型控制服务.producer_exec(self.producer,self.request_topic,req_params,'上传模型')
msg=Biz_模型控制服务.consumer_exec(self.consumer,self.response_topic,'上传模型')
Biz_模型控制服务.py
def producer_exec(producer:KafkaProducer,req_topic:str,req_params:dict,action_chn:str):
producer.send(req_topic,value=req_params)
logging_kafka_req(action_chn,req_topic,req_params)
def consumer_exec(consumer:KafkaConsumer,res_topic:str,action_chn:str):
msg={}
for message in consumer:
msg=json.loads(message.value.decode())
logging_kafka_res(action_chn,res_topic,msg)
break
consumer.commit()
time.sleep(2)
return msg
def kafka_producer_init(host:str=None)
kafka_host=host
if kafka_host is None:
kafka_host = env_c.get('kafka_host')
producer = KafkaProducer(
bootstrap_servers=[kafka_host],
key_serializer=lambda k: json.dumps(k).encode(),
value_serializer=lambda v: json.dumps(v).encode(),
)
return producer
def kafka_consumer_init(res_topic:str,time_out:int,host:str=None):
kafka_host=host
if kafka_host is None:
kafka_host=env_c.get('kafka_host')
consumer=KafkaConsumer(
res_topic,
bootstrap_servers=kafka_host,
group_id='cloud-chm',
consumer_timeout_ms=time_out
)
topic_partition=consumer.assignment()
partitions=consumer.beginning_offsets(list(topic_partition))
if partitions != {}:
for message in consumer:
pass
consumer.commit()
return consumer
def logging_kafka_req(api_name:str,request_topic:str,request_params:dict):
logging.info(f'{api_name} request topic:{request_topic}')
logging.info(f'{api_name} request params:{request_params}')
def logging_kafka_res(api_name:str,response_topic:str=None,response_result:dict=None):
logging.info(f'{api_name} response topic:{response_topic}')
logging.info(f'{api_name} response result:{response_result}')
标签：producer,使用,kafka,topic,host,str,自动化,Kafka,consumer From： https://www.cnblogs.com/nullnullnull/p/17532957.html