长春整站优化,网络服务器分类,ai生成网页设计,wordpress 本机安装通过assign、subscribe两者之一为消费者设置消费的主题consumer KafkaConsumer(bootstrap_servers[127.0.0.1:9092],auto_offset_resetlatest,enable_auto_commitTrue, # 自动提交消费数据的offsetconsumer_timeout_ms 10000, # 如果1秒内kafka中没有可供消费的数据#xff0…通过assign、subscribe两者之一为消费者设置消费的主题consumer KafkaConsumer(bootstrap_servers[127.0.0.1:9092],auto_offset_resetlatest,enable_auto_commitTrue, # 自动提交消费数据的offsetconsumer_timeout_ms 10000, # 如果1秒内kafka中没有可供消费的数据自动退出value_deserializerlambda m: json.loads(m.decode(ascii)), #消费json 格式的消息client_idconsumer-python3)# consumer.assign([TopicPartition(MY_TOPIC1, 0)])# msg next(consumer)# print(msg)consumer.subscribe(MY_TOPIC1)for msg in consumer:print (msg)API及常用参数说明class kafka.KafkaConsumer(*topics, **configs)*topics (str) – 可选设置需要订阅的topic如果未设置需要在消费记录前调用subscribe或者assign。client_id (str) – 客户端名称默认值: ‘kafka-python-{version}’group_id (str or None) – 消费组名称。如果为None则通过group coordinator auto-partition分区分配offset提交被禁用。默认为Noneauto_offset_reset (str) – 重置offset策略: earliest将移动到最老的可用消息, latest将移动到最近消息。 设置为其它任何值将抛出异常。默认值latest。enable_auto_commit (bool) – 如果为True将自动定时提交消费者offset。默认为True。auto_commit_interval_ms (int) – 自动提交offset之间的间隔毫秒数。如果enable_auto_commit 为true默认值为 5000。value_deserializer(可调用对象) - 携带原始消息value并返回反序列化后的valuesubscribe(topics(), patternNone, listenerNone)订阅需要的主题topics (list) – 需要订阅的主题列表pattern (str) – 用于匹配可用主题的模式即正则表达式。注意必须提供topics、pattern两者参数之一但不能同时提供两者。metrics(rawFalse)获取消费者性能指标。参考APIhttps://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html原文强参考https://www.cnblogs.com/shouke/p/10463377.html
