DSPRiverInspection/producer.py

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json
import time
import random,string
def producer_demo():


    cnt_online=1;cnt_offline=0
    Tecent=False;
    
    #topic_on='thsw';topic_off='thsw2';
    #server=['212.129.223.66:19092'];
    server=["192.168.11.242:9092"]
    #server=['101.132.127.1:19092']

    topic_on='dsp-alg-online-tasks';topic_off='dsp-alg-offline-tasks'
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    producer = KafkaProducer(
        bootstrap_servers=server,#tencent yun 
        value_serializer=lambda v: v.encode('utf-8'))
    # 发送三条消息
   
    if not Tecent:
        #pull_channel = "rtmp://live.play.t-aaron.com/live/THSA"
        #push_channel = 'rtmp://live.push.t-aaron.com/live/THSB'
        #pull_channel = 'rtmp://live.play.t-aaron.com/live/THSAa_hd'
        pull_channel = 'http://live.play.t-aaron.com/live/THSAa_hd.m3u8'
        push_channel = "rtmp://live.push.t-aaron.com/live/THSBa"
    else:
        pull_channel = "rtmp://demoplay.yunhengzhizao.cn/live/THSA_HD5M"
        push_channel = "rtmp://127.0.0.1:1935/live/test"
        #pull_channel = 'rtmp://live.play.t-aaron.com/live/THSAa'
        #push_channel = 'rtmp://127.0.0.1:1975/live/test'
    for i in range(0, cnt_online):
        time.sleep(0.0001)
        #''.join(random.sample(string.ascii_letters ,16)  )
        msg_dict = {                   
                    "request_id":'nn'+''.join(random.sample(string.ascii_letters ,30)  ) ,
                    "models":[
                                { "id":"0","config":{}},
                                { "id":"1","config":{}},
                                { "id":"2","config":{}},
                                { "id":"3","config":{}},
                                
                            ],
                    "pull_url":pull_channel,      
                    "push_url":push_channel,                    
                    "results_base_dir": "XJRW202111291703"+str(random.randint(10,99)),
                    }
        msg = json.dumps(msg_dict)
        #msg = msg_dict
        future = producer.send(
            topic_on,
            #key='count_num',  # 同一个key值，会被送至同一个分区
            msg
            )


        print('online send {}'.format(str(i)))
        try:
            future.get(timeout=10) # 监控是否发送成功           
        except kafka_errors:  # 发送失败抛出kafka_errors
            traceback.format_exc()

    for i in range(0, cnt_offline):        
        msg_dict = {
                    
                    "request_id":'bb'+''.join(random.sample(string.ascii_letters ,30)  ) ,
                    "models":[
                               { "id":"0","config":{}},
                               { "id":"1","config":{}},
                               { "id":"2","config":{}},
                               { "id":"3","config":{}},
                            ],
                    "original_url":"http://vod.play.t-aaron.com/customerTrans/c49a2c620795d124f2ae4b10197b8d0e/303b7a58-17f3ef4494e-0004-f90c-f2c-7ec68.mp4",
                    "original_type":"mp4",
                    "results_base_dir": "XJRW202203171535"+str(random.randint(10,99)),
                    }
                    
       
    
        msg = json.dumps(msg_dict)
        #msg = msg_dict
        future = producer.send(topic_off,msg)
        print('offline send {}'.format(str(i)))
        try:
            future.get(timeout=10) 
        except kafka_errors:
            traceback.format_exc()

if __name__=='__main__':
    producer_demo()