tencent cloud

 producer = KafkaProducer(
         bootstrap_servers = ['xx.xx.xx.xx:port'],#地址
         api_version = (1, 1),
         security_protocol = "SASL_PLAINTEXT",
         sasl_mechanism = "PLAIN",
         sasl_plain_username = "username",#用户名
         sasl_plain_password = "password",#密码
 )

 message = "Hello World! Hello Ckafka!"
 msg = json.dumps(message).encode()
 producer.send('topic_name', value = msg)#topic名称
 print("produce message " + message + " success.")
 producer.close()

   consumer = KafkaConsumer(
       'topic_name',#topic名称
       group_id = "group_id",#消费组
       bootstrap_servers = ['xx.xx.xx.xx:port'],#地址
       api_version = (1,1),

       security_protocol = "SASL_PLAINTEXT",
       sasl_mechanism = 'PLAIN',
       sasl_plain_username = "username",#用户名
       sasl_plain_password = "password",#密码
   )

   for message in consumer:
       print ("Topic:[%s] Partition:[%d] Offset:[%d] Value:[%s]" %
       (message.topic, message.partition, message.offset, message.value))