Python 多线程编写 Kafka Producer

心已赠人 2022-01-10 13:57 763阅读 0赞

**多线程生产到同一 topic**

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    import random
    import time
    import threading
    from kafka import KafkaProducer
    from traitlets import log
    
    
    def on_send_success(record_metadata):
        print(record_metadata.topic)
        print(record_metadata.partition)
        print(record_metadata.offset)
    
    
    def on_send_error(excp):
        log.error('I am an errback', exc_info=excp)
        # handle exception
    
    def send_messag(topic, id, interval_time):
    
        producer = KafkaProducer(bootstrap_servers=['cdh01:9092', 'cdh02:9092', 'cdh03:9092'])
        while True:
            startTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
            message = "|".join((id, startTime, str(interval_time)))
    
            producer.send(topic, message.encode("utf-8")).add_callback(on_send_success).add_errback(on_send_error)
            # 间隔多长时间生产一条消息
            time.sleep(interval_time)
    
    interval_times = [60, 300, 600, 1200, 1800, 3600]
    
    #开启50个线程
    for i in range(0, 50):
        deviceId = "%06d" % random.randint(0, 999999)
    
        interval_time = times[random.randint(0, len(interval_times) - 1)]
    
        recv_thread = threading.Thread(target=send_messag, args=("my-topic", id, interval_time))
        
        # 守护主线程，主线程退出后子线程直接销毁
        recv_thread.setDaemon(True)
        recv_thread.start()
    
    while True:
        #主线程一直运行
        time.sleep(5000)

发表评论取消回复

表情：

评论列表（有 0 条评论，763人围观）

还没有评论，来说两句吧...

相关阅读

相关 kafka生产者producer

java Client：元数据： ![20210625093504754.png][] public final class Cluster {

迷南。/ 2023年02月20日 02:13/ 0 赞/ 65 阅读

相关 java kafka producer_Java实现Kafka Producer方法

import org.apache.kafka.clients.producer.\; import java.util.Properties; public class

﹏ヽ暗。殇╰゛Y/ 2022年11月03日 11:15/ 0 赞/ 190 阅读

相关 java kafka 多线程消费

我们先来看下简单的kafka生产者和消费者模式代码：生产者KafkaProducer / @author xiaofeng @version V1.0 @t

待我称王封你为后i/ 2022年05月28日 11:22/ 0 赞/ 229 阅读

相关 python多线程

单线程: from time import ctime,sleep def music(): for i in range(2):

男娘i/ 2022年04月13日 05:11/ 0 赞/ 330 阅读

相关 python多线程

现在的CPU都是多核的，但是我们写的程序大部分都是单核的，为了缩短跑特征的时间，学习下python3的多线程。 python提供了两个模块来实现多线程thread 和thre

喜欢ヅ旅行/ 2022年02月09日 11:34/ 0 赞/ 354 阅读

相关 Kafka之Producer

producer 发布消息 1、内容将数据转化为“消息”，并通过网络发送给Broker，每条数据被称为“消息”，每条信息表示为一个三元组，即<t

た入场券/ 2022年02月05日 03:13/ 0 赞/ 216 阅读

相关 Kafka 多种语言编写 Producer

Java 编写 Producer maven 依赖 <dependency> <groupId>org.apache.kafka</groupId>

末蓝、/ 2022年01月10日 14:45/ 0 赞/ 243 阅读

相关 Python 多线程编写 Kafka Producer

多线程生产到同一 topic !/usr/bin/env python -- coding: utf-8 -- import random

心已赠人/ 2022年01月10日 13:57/ 0 赞/ 764 阅读

相关 Kafka幂等Producer和事务Producer

一、Kafka 消息交付可靠性 Kafka 对Producer和Consumer 需要处理的消息提供三种承诺：最多一次（at most once）：消息可能丢失

た入场券/ 2021年11月11日 13:44/ 0 赞/ 352 阅读

相关 Kafka Producer配置

java生产者配置： <table> <thead> <tr> <th>NAME</th> <th>DESCRIPTION</th> <th

ゞ浴缸里的玫瑰/ 2021年06月22日 15:36/ 0 赞/ 454 阅读