Kafka Producer(Python threading)

import threadingimport timeimport randomfrom kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=‘192.168.1.10:9092‘)threads = []

class MyThread(threading.Thread):    def __init__(self, threadName, delay):        threading.Thread.__init__(self)        self.threadName=threadName        self.delay=delay

def run(self):        sendinfo(self.threadName, self.delay)

def sendinfo( threadName, delay):   count = 0   while count < 5:      time.sleep(delay)      count += 1      data = "".join(random.sample(          [‘a‘, ‘b‘, ‘c‘, ‘d‘, ‘e‘, ‘f‘, ‘g‘, ‘h‘, ‘i‘, ‘j‘, ‘k‘, ‘l‘, ‘m‘, ‘n‘, ‘o‘, ‘p‘, ‘q‘, ‘r‘, ‘s‘, ‘t‘, ‘u‘, ‘v‘,           ‘w‘, ‘x‘, ‘y‘, ‘z‘], 10)).replace(" ", "")

word=("%s, %s, %s, %s" % (threadName, count, data, time.ctime(time.time())))      producer.send(‘test‘, key=threadName, value=word)      print (word)

try:    t1=MyThread("Thread-1",0)    threads.append(t1)    t2=MyThread("Thread-2",0)    threads.append(t2)    t3=MyThread("Thread-3",0)    threads.append(t3)

for t in threads:        t.start()

for t in threads:        t.join()

producer.send(‘test‘, key="Thread-1", value="exit")    producer.send(‘test‘, key="Thread-2", value="exit")    producer.send(‘test‘, key="Thread-3", value="exit")

print ("exit program with 0")except:   print ("Error: failed to run producer program")
时间: 2024-11-05 13:43:41

Kafka Producer(Python threading)的相关文章

Kafka Consumer(Python threading)

import threadingfrom kafka import KafkaConsumer threads = [] class MyThread(threading.Thread): def __init__(self, threadName, keyName): threading.Thread.__init__(self) self.threadName=threadName self.keyName=keyName def run(self): receiveinfo(self.th

python threading模块使用 以及python多线程操作的实践(使用Queue队列模块)

今天花了近乎一天的时间研究python关于多线程的问题,查看了大量源码 自己也实践了一个生产消费者模型,所以把一天的收获总结一下. 由于GIL(Global Interpreter Lock)锁的关系,纯的python代码处理一般逻辑的确无法活动性能上的极大提升,但是在处理需要等待外部资源返回或多用户的应用程序中,多线程仍然可以作为一个比较好的工具来进行使用. python提供了两个模块thread和threading 来支持python的多线程操作.通俗的讲一般现在我们只使用threading

kafka+docker+python

昨天晚上刚刚才花3小时看完<日志:每个软件工程师都应该知道的有关实时数据的统一概念>. 今天就把kafka在docker容器里运行起来,github上有几个,但都太复杂了. 我自己写个最简单的python的demo体验一下:https://github.com/xuqinghan/docker-kafka 和上周部署taiga相比,kafka不愧是大家手笔,基本无坑,简单记录一下: 首先是docker-compose.yml version: '3.1' services: zoo: imag

Kafka 通过python简单的生产消费实现

使用CentOS6.5.python3.6.kafkaScala 2.10  - kafka_2.10-0.8.2.2.tgz (asc, md5) 一.下载kafka 下载地址 https://kafka.apache.org/downloads 里面包含zookeeper 二.安装Kafka 1.安装zookeeper mkdir /root/kafka/ tar -vzxf kafka_2.10-0.8.2.2 cd /root/kafka/kafka_2.10-0.8.2.2 cat 

kafka producer源码

producer接口: /** * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this

Kafka Producer相关代码分析

Kafka Producer相关代码分析 标签(空格分隔): kafka Kafka Producer将用户的消息发送到Kafka集群(准确讲是发送到Broker).本文将分析Producer相关的代码实现. 类kafka.producer.Producer 如果你自己实现Kafka客户端来发送消息的话,你就是用到这个类提供的接口来发送消息.(如果你对如何利用Producer API来发送消息还不是很熟悉的话,可以参看官方的例子).这个类提供了同步和异步两种方式来发送消息. 异步发送消息是基于同

[Python]threading local 线程局部变量小测试

概念 有个概念叫做线程局部变量,一般我们对多线程中的全局变量都会加锁处理,这种变量是共享变量,每个线程都可以读写变量,为了保持同步我们会做枷锁处理.但是有些变量初始化以后,我们只想让他们在每个线程中一直存在,相当于一个线程内的共享变量,线程之间又是隔离的.python threading模块中就提供了这么一个类,叫做local. 多线程中共享变量和局部变量的区别我画两个小图,简单描述下(作图能力一般,请见谅,概念性的东西大家可以google下,很多好文章) 全局变量 线程局部变量 对比: 下面是

kafka producer生产数据到kafka异常:Got error produce response with correlation id 16 on topic-partition...Error: NETWORK_EXCEPTION

kafka producer生产数据到kafka异常:Got error produce response with correlation id 16 on topic-partition...Error: NETWORK_EXCEPTION 1.问题描述 2017-09-13 15:11:30.656 o.a.k.c.p.i.Sender [WARN] Got error produce response with correlation id 25 on topic-partition t

kafka producer consumer

package demo; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; public class producer { private final Producer<String, String> producer; public final static