04_kafka python客户端_Producer模拟

"""
Kafka Producer Test
using kafka-python library
"""
# -*- encoding: utf-8 -*-
# Author: shayzhang@sina.com

# import KafkaProducer class
from kafka import KafkaProducer
# import KafkaError class
from kafka.errors import KafkaTimeoutError
# time for message timestamp
import time


def main():
    # 创建producer实例，并传入bootstrap_servers列表(brokers), 修改producer实例配置
    producer = KafkaProducer(bootstrap_servers=["192.168.229.100:9092", "192.168.229.101:9092", "192.168.229.102:9092"])

    # topic to be published
    topic = ‘ctopic‘

    # message value to be published, must be bytes type
    msg = bytes(‘hello_from_python‘, encoding=‘utf-8‘)
    # for python2:  msg = b‘hello_from_python‘

    # message key, must be bytes type
    # used to determine which partition the message will be stored
    key = bytes(‘shay‘, encoding=‘utf-8‘)
    # for python2:  key = b‘shay‘

    # Async send, default
    try:

        # get partitions for the topic
        partition_set = producer.partitions_for(topic)
        for e in partition_set:
            print("Partition: " + str(e))
            # print ‘Partition: ‘+ str(e)

        future = producer.send(topic, msg, key, partition=None, timestamp_ms=time.time())
        # block until all records are sent to cluster
        producer.flush()

        print("Message Send!")
        # print "Message send!"
    except  KafkaTimeoutError:
        print("Kafka Timeout")
        # print("Kafka Timeout")


if __name__ == ‘__main__‘:
    main()