1回答
我正在定制OpenShot来改变视频的亮度。我已经知道需要将sws_getColorspaceDetails / sws_setColorspaceDetails函数添加到producer_avformat的convert_image()中。
但我想知道如何从python代码访问producer_avformat对象以将亮度作为参数传递。
在Python中，代码生成器在这里创建: self.p = mlt.Producer( self.profile，'xml:%s‘% self.file_name)
但它是另一个生产者。
问候
浏览 34提问于2016-04-27得票数 0
1回答
我试图复制中给出的步骤。在尝试时，给出了一个Kafka Consumer和Kafka Producer python代码，我可以在python交互式终端上运行这些代码，并且消费者控制台能够给出输出，但是如果我将它们传递到python (*.py)中，它就不会消耗任何东西。
消费者
from kafka import KafkaConsumer
consumer = KafkaConsumer('sample')
for message in consumer:
print (message)
生产者
from kafka import KafkaProducer
pro
浏览 1提问于2019-05-06得票数 1
回答已采纳
我是python的初学者，不知道如何做到这一点:我正在运行一个python脚本，它每5-10秒将一个新值放入列表中。我想从另一个多线程python脚本中的列表中选择这些元素，但是每个线程都有一个值，所以不应该重用一个值，如果没有下一个值，那么等待下一个值出现。我有一些代码试图做，但没有成功：
创建值的脚本：
values = ['a','b','c','d','e','f']
cap = []
while True:
cap.append(random.choice(values))
浏览 1提问于2018-07-02得票数 0
回答已采纳
我有一个下面的python代码，它在python - python PUB/SUB ZeroMQ ( pyzmq )场景中成功工作。
我可以通过c++订阅这个python服务器吗？
我还添加了一个c++客户端代码，但它不起作用。
Python：
def send_array_and_str(socket, img, string, flags=0):
global count
print(count , " sended")
## numpy array gönderirken shape bilgilerini de msg olarak eklem
浏览 6提问于2020-04-17得票数 1
3回答
我有写在Java上的代码，它将所有数据写入文件，然后我有python脚本来处理这些数据。
它们完全分开运行，python脚本可以按计划运行，但它也从文件中删除了处理过的记录。问题是，在实现对文件的访问时，第一个进程中的java代码将尝试编写内容，而第二个进程中的python代码将尝试删除已处理的记录？
首先想到的是在某个进程更新文件时物理创建.lock文件，但也许还有其他解决方案可以考虑？
谢谢。
浏览 3提问于2013-03-27得票数 1
在使用者行中获取消息中的错误
代码：
使用者=KafkaConsumer(主题，group_id=groupid，bootstrap_servers=host)用于使用者中的消息：
错误:关于使用者中的消息:文件"\Python\Python38-32\lib\site-packages\kafka\consumer\group.py"，第1192行，next返回self.next_v2() self.next_v2行，next_v2返回下一步(self._iterator)文件"\Python\Python38-32\lib\site-packages\kafka\c
浏览 4提问于2020-05-15得票数 0
1回答
我有多个编写量大的Python应用程序(producer1.py，producer2.py，.)我希望将异步、非阻塞写入器(consumer.py)实现为一个单独的进程，这样生产者就不会被磁盘访问或争用所阻塞。
为了使这更容易优化，假设我只需要公开一个日志调用，该调用将一个固定长度的字符串从生产者传递给编写器，并且编写的文件不需要按调用时间排序。目标平台可以是Linux。我应该如何在调用线程上以最小的延迟代价来实现这一点？
这似乎是多个无锁SPSC队列的理想设置，但我找不到任何Python实现。
编辑1
我可以在/dev/shm上将循环缓冲区实现为内存映射文件，但我不确定是否会在Python
浏览 4提问于2017-11-29得票数 0
回答已采纳
1回答
我正在尝试了解Rest代理解决方案是否可以在我的解决方案中帮助我。我评论说： 我正在尝试从我的前端(react)发送一个图像，以响应由我的应用程序的客户端生成的事件。该图像被发送给制作人卡夫卡，卡夫卡将派他的消费者在几种机器学习模型中处理它，并将响应返回给该制作人。 流程： 前端(img) -> ApiRest (Spring Boot) ->生产者->消费者(Python) ->向客户端返回响应。 我知道我需要Kafka作为一个请求-响应，将响应返回给发出事件的客户端。 我找对地方了吗？ 另一方面，我想知道在同一主题中使用不同语言(Java和Python)的几个使用
浏览 26提问于2019-04-08得票数 0
我是kafka的新手，我只是在尝试Hortonworks Sandbox中的一个示例程序。获取以下错误。任何帮助都将不胜感激。
from kafka import KafkaProducer
from kafka.errors import KafkaError
producer = KafkaProducer(bootstrap_servers=['sandbox.hortonworks.com:9092'])
topic = "kafkatopic"
producer.send(topic, b'test message')
错误：
Tr
浏览 0提问于2017-03-11得票数 1
我有一个运行并经过测试的Kafka集群，并且正在尝试使用Python脚本向代理发送消息。当我使用Python3 shell并调用producer方法时，这是可行的，但是当我将这些相同的命令放入Python3并执行它时--脚本似乎挂起了。
我正在为消费者和生产者使用kafka-python库。当我使用Python3 shell时，我可以看到使用Kafka工具2.0.4出现在主题中的消息--我在python代码中尝试过各种循环和语句，但似乎没有什么能使它“运行”完成。
>>>from kafka import KafkaProducer
>>>producer
浏览 0提问于2019-04-05得票数 1
回答已采纳
1回答
我正在尝试使用multiprocessing在Python中创建一个简单的生产者/消费者模式。它可以工作，但它挂在poll.join()上。
from multiprocessing import Pool, Queue
que = Queue()
def consume():
while True:
element = que.get()
if element is None:
print('break')
break
print('Consumer closing&
浏览 1提问于2016-02-23得票数 1
回答已采纳
2回答
我正在使用Python库发送一个POST请求。程序中生成POST数据的部分可以将写入一个类似文件的任意对象(输出流)。
我怎样才能使这两个部分合身呢？
我本以为requests会为这个用例提供一个流接口，但它似乎没有，它只接受一个类似文件的对象作为data参数，从中读取。它没有提供一个类似文件的对象，我可以在其中编写。
这是Python库的一个基本问题吗？
到目前为止的想法：
似乎最简单的解决方案是使用fork()，并让请求库通过管道与POST数据生成器进行通信。
有更好的办法吗？
或者，我可以尝试使POST数据生产者复杂化。但是，这是解析一个XML流(来自stdin)并生成一个新的XML流作
浏览 3提问于2016-10-13得票数 7
回答已采纳
我最近尝试使用python向Kafka发送消息。当使用简单的字节消息时，它可以工作。但是现在，我有了一个json数据，我需要将其发送到Kafka主题，然后Java应用程序将使用它。
我试图找出如何将json转换为byteArray (这是Java应用程序期望的有效负载)。所以，我想出了下面的python脚本。但它失败了，因为json中有一些布尔变量，并且我得到了一个类型错误，因为Json true和Python True的大小写不同。我尝试用单引号将json括起来，但再次收到错误消息'EOL while scanning string literal‘。只有我修复了这个错误，我才能知道
浏览 95提问于2018-07-21得票数 0
我正在学习如何在python下使用kafka。
我让kafka服务器在运行，并且有一个正在监听mytopic的消费者。
现在，我使用confluent-kafka编写了这段代码
from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('mytopic', key='hello', value='world')
当我运行python并粘贴代码时，它可以工作(消费者正在收集)
浏览 16提问于2018-01-26得票数 0
回答已采纳
我目前正在使用tweepy lib和python (2.7.3)对我的项目使用twitter流API。我正在接收的数据的速率非常高。有没有可能让它慢下来？我在某处读到，Twitter不喜欢将流媒体延迟太久的想法。
我想做的另一件事是引入数据，比方说5分钟，5分钟后数据应该被排序(基于任何事情！例如散列标签的字数等)。但我认为这会减慢我的流媒体速度(因为我使用的是一个进程)。
那么，你能建议一种处理它的方法吗？多线程？多进程？
浏览 1提问于2013-02-27得票数 0
在Python语言中，多处理中的队列和JoinableQueue有什么不同？这个问题已经被问到了here，但正如一些评论指出的那样，被接受的答案没有什么帮助，因为它所做的一切就是引用文档。有人能解释一下什么时候使用一个和另一个有什么区别吗？例如，如果除了提供两个额外的方法join()和task_done()之外，JoinableQueue几乎是相同的东西，那么为什么要选择使用Queue而不是JoinableQueue。此外，我链接的帖子中的另一个答案提到了Based on the documentation, it's hard to be sure that Queue is ac
浏览 22提问于2020-07-14得票数 1
回答已采纳
我最近一直在学习Python，当我探索一些关于threading.Thread的知识时，我遇到了一些问题。示例代码如下：
#!/usr/bin/env python3.0
# -*- coding: utf-8 -*-
#!/usr/bin/env python3.0
# -*- coding: utf-8 -*-
from threading import Condition, Thread
import time
class ProductAndConsumer(object):
""" test the threading.Condit
浏览 3提问于2014-04-18得票数 0
回答已采纳
1回答
我有几台从Python生成xml文件的服务器，还有一些使用Java使用这些xml的其他服务器。我最近才研究过JMS和，并决定尝试使用它来传递xml文件。
因此，我在使用者服务器上设置了ActiveMQ守护进程，并认为我应该在产品上实现一些循环方法，以便在使用者之间均匀地分发xmls。
Python -----------------------------> ActiveMQ ---> Java
\ /
\ /
--------
浏览 2提问于2009-03-16得票数 2
回答已采纳
如何使用kafka-python发送自定义负载？ 我有“两台ubuntu机器”，而且都在同一个WIFI网络下(一个地址是172.20.10.2，另一个地址是172.20.10.7)，我可以使用深流test4 python脚本通过kafka成功地传输检测到的包围盒信息。但我想要定制的有效载荷..。因此，我尝试了一些kafka-python scrpit。对于生产者： from time import sleep
from json import dumps
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_s
浏览 11提问于2021-07-29得票数 0
我正在尝试创建一个程序，它启动一个进程池，比如5个进程，执行一些操作，然后退出，但让5个进程处于打开状态。稍后，用户可以再次运行该程序，而不是启动新进程，而是使用现有的5。基本上，它是生产者-消费者模型，其中：
生产者的数量各不相同。
消费者的数量是不变的。
生产者可以在不同的时间，由不同的程序，甚至不同的用户。
我正在使用内置的multiprocessing模块(目前在Python2.6.4中)，但目的是最终迁移到3.1.1。
下面是一个基本的使用场景：
开始状态-没有进程正在运行。
用户启动program.py operation -一个生产者，五个消费者正在运
浏览 3提问于2010-01-08得票数 9
3回答
我有一个数据生产者和一个数据消费者。生产者异步生产，反过来，当有数据要消费时，我希望消费者异步消费。 我立即想到的解决这个问题的方法是使用某个队列对象，该对象具有可等待的shift/get、much like this async queue in the python standard 但是，我搜索了一下，找不到任何具有这种数据结构的JS库供我使用。我本以为这是一种常见的模式。 在JS中解决这个问题的常见模式是什么，有没有什么库可以提供帮助？
浏览 148提问于2019-06-27得票数 4
回答已采纳
1回答
正如我们所知道的，我们可以向kafka producer发送一个密钥，该密钥在内部进行了哈希处理，以找到主题数据中的哪个分区。我有一个生产者，在那里我发送一个JSON格式的数据。 kafka-console-producer --broker-list 127.0.0.1:9092 --topic USERPROFILE << EOF
{"user_id" : 100, "firstname":"Punit","lastname":"Gupta", "countrycode":
浏览 23提问于2019-12-16得票数 2
回答已采纳
这个问题与Python 有关。我要求为我的特定场景提供一个合适的进程间通信数据结构：
我的情景
我有一个制片人和一个消费者。
制片人每隔10分钟就生产一只相当小的熊猫数据，然后把它放到python.multiprocess.queue上。
使用者是一个GUI轮询，每100 is就轮询一次python.multiprocess.queue。消费者捕获生产者生产的每一个DataFrame是非常关键的。
我的思想
python.multiprocess.queue服务的目的(我认为)，并令人惊讶地简单使用！(赞美绿色的领主！)但是，，我显然没有充分利用队列的潜力，只有一个生产者、一个
浏览 0提问于2016-04-20得票数 4
回答已采纳
我用Python/Twisted编写了一个音频广播服务器。它工作得很好，但是内存使用增长太快了！我认为这是因为一些用户的网络可能不够好，无法及时下载音频。
我的音频服务器将音频数据广播到不同听众的客户端，如果他们中的一些人不能及时下载音频数据，也就是说，我的服务器一直保存着音频数据，直到听众收到。更重要的是，我的音频服务器是一个广播服务器，它接收音频数据，并将它们发送到不同的客户端，我认为Twisted将这些数据复制到不同的缓冲区中，即使它们是相同的音频片段。
我希望减少内存使用量，因此我需要知道客户端何时接收到音频，以便我可以决定何时丢弃一些速度较慢的客户端。但我不知道如何使用Twisted
浏览 2提问于2009-11-09得票数 2
回答已采纳
2回答
我正在尝试读取管道中的行并对其进行处理，但我正在做一些愚蠢的事情，并且我不知道是什么。生产者将无限期地继续生产生产线，如下所示：
producer.py
import time
while True:
print 'Data'
time.sleep(1)
消费者只需要定期检查线路：
consumer.py
import sys, time
while True:
line = sys.stdin.readline()
if line:
print 'Got data:', line
else:
浏览 1提问于2009-06-08得票数 15
回答已采纳
场景
在我的应用程序中，有3个进程正在各自文件夹中的共享驱动器上复制文档。
一旦在共享驱动器上复制了任何文档(由任何进程复制)，目录监视程序(Java)代码就会提取文档并使用"Process“调用Python脚本，并对文档进行一些处理。代码片段如下：
Process pr = Runtime.getRuntime().exec(pythonCommand)；//检索python脚本BufferedReader bfr =新的BufferedReader(新的InputStreamReader(pr.getInputStream()的输出；String line = "&
浏览 0提问于2019-10-18得票数 4
回答已采纳
1回答
我正在用Twisted编写一个自定义的SSL代理。我总是遇到一个经常发生的问题，但我不知道问题是什么。 当我尝试通过与twisted.protocols.portforward函数完全一样的registerProducer函数将客户机传输连接到服务器的传输时，我一直收到这个错误。 File "/opt/Memory/Mobile/Proxy/forwarder.py", line 40, in connectionMade
self.peer.transport.registerProducer(self.transport, True)
浏览 19提问于2021-01-11得票数 1
1回答
当我试图将一些消息发布到代理时，我得到了以下错误：
Exception in thread Thread-4:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/threading.py", line 932, in _bootstrap_inner
self.run()
File "/usr/local/lib/python3.8/site-packages/opcua/client/client.py", line 66, in run
self.cli
浏览 13提问于2020-05-18得票数 0
1回答
我正在使用Apache Spark Streaming，使用TCP连接器来接收数据。我有一个连接到传感器的python应用程序，并创建了一个TCP服务器，该服务器等待来自Apache Spark的连接，然后通过此套接字发送json数据。
我如何才能加入多个独立的传感器源，将数据发送到Apache Spark上的同一接收器？
浏览 3提问于2017-04-16得票数 1
1回答
我有两个关于python中的线程和队列的问题。
max_size arg在queue.Queue()中做什么？
根据线程的#(num_worker_threads)，是否有任何性能改进？我找不到任何改进。
-> --如果没有依赖于线程#的改进，我们为什么需要这样做？
import time
import queue
import threading
num_worker_threads = 10
store = []
def worker():
while True:
item = q.get()
if item is Non
浏览 2提问于2017-10-31得票数 0
1回答
我找不到任何答案：和有什么区别？
选择其中任何一个有什么好处或缺点吗？
据我所知：
Kafka是用Java编写的，Kafka-python是Python客户端，用于与"Java流“<code>e 213</code>通信。
Faust是一个纯"Python流“
所以，如果我计划只使用Python，那么Faust应该是更好的选择，如果我想拥有更广泛的兼容性()，那么应该使用Kafka +Kafka吗？
注:我对卡夫卡的使用并不熟悉，我试图理解不同解决方案的利弊。
我将非常感谢您的任何建议！
浏览 0提问于2021-01-31得票数 5
回答已采纳
所以我从线程模块中使用python的条件：
from threading import Thread, Condition
condition = Condition()
我有一个生产者类(Thread的子类)，它本质上在for循环中将项添加到队列中，直到队列满(即达到定义的最大长度)和使用者类，除非队列是空的，否则在for循环中会弹出项。在生产者中，如果队列已满，则有一个condition.wait()语句，类似地，在使用者类中，如果队列为空，则有一个condition.wait()语句。如果这些条件都没有满足(队列既不满也不空)，那么每个类都会执行它的任务(分别为生产者或消费者向队列添加一
浏览 3提问于2017-09-20得票数 0
我在用python做一个简单的任务时遇到了一些麻烦，请看一下这个问题，看看你能不能帮我。
复制步骤
我使用了这个代码：
import faust
from settings import KAFKA_SERVER
app = faust.App('streams', broker=KAFKA_SERVER, producer_acks=0, store='rocksdb://')
class ProjetoRateio(faust.Record):
codigoProjeto: str
combinacao: str
grade:
浏览 3提问于2021-05-24得票数 0
我正在尝试用
json.loads(json_string)
但是它返回一个字符串而不是一个dict。我可以通过再次解析得到预期的结果。
json.loads(json.loads(json_string))
但我不明白为什么。
我从web钩子接收到一个字节对象：
bytes_object = b'"{\\"action\\":\\"connection_test\\",\\"data\\":{}}"'
然后字节对象被解码为utf-8：
decoded_bytes = bytes_object.decode(
浏览 2提问于2021-05-19得票数 1
1回答
我真的很喜欢将操作应用到流IO源的管道/管道的概念。我对构建处理非常大的日志文件的工具很感兴趣。从Python/Ruby迁移到Haskell的一个吸引人的地方是更容易编写并行代码，但我找不到任何有关这方面的文档。我如何设置一个从文件中读取行并并行处理它们的管道流(即。对于8个核心，它应该读取8行代码，并将它们传递给8个不同的线程进行处理，然后重新收集(等等)，理想情况下尽可能少地“仪式”……
可选的是，是否需要按顺序重新连接这些行，如果这会影响过程的速度？
我确信可以使用Parallel Haskell一书中的想法自己拼凑出一些东西，但在我看来，在管道工作流中间并行运行一个纯函数(parmap
浏览 5提问于2014-11-05得票数 6
下午好，我最近才开始和卡夫卡合作，我有一个关于制片人的问题。
最初，我试图在C#中构建一个没有模式的简单生产者。到目前为止，这是工作的，代码也提供了一个简短的版本。
没有模式的生产者代码：
var config = new ProducerConfig
{
BootstrapServers = "localhost:9092",
BrokerAddressFamily = BrokerAddressFamily.V4,
};
using (var producer = new ProducerBuilder<Null, string>(config
浏览 0提问于2020-10-21得票数 0
1回答
在从下面的学习线程和队列时，我构建了以下代码。
from __future__ import print_function
import queue
import threading
import time
class a3_4(threading.Thread):
q = queue.Queue()
def __init__(self, begin, end, waiting_time):
self.begin = begin
self.end = end
self.waiting_time = waiting_time
浏览 5提问于2015-11-12得票数 2
回答已采纳
我有一个Python程序，它产生多个生产者线程，然后有一个循环等待Queue对象在其中有一些东西。它看起来像这样
for t in threads:
t.start()
while len(threads):
if not queue.empty():
response = queue.get()
# handle response
queue.task_done()
else:
sleep(1)
threads = [t for t in threads if t.is_alive()]
必须有一
浏览 1提问于2015-06-11得票数 2
我正在Python项目中工作。
它的功能之一是有人从web应用程序中发送一些数据，将数据提前发送到移动设备。
对于此功能，应用程序使用第三方服务作为和。以下是应用程序的数据流。
Web应用
由于Amazon端可能存在网络延迟、操作延迟和故障等问题，web应用程序使用作为调度器应用程序来实现异步、重试等行为，并将SNS发送任务存储在队列中并进行异步处理。
芹菜作为调度器使用RabbitMQ作为它的代理应用程序。以下是数据流。
Web >芹菜-> RabbitMQ -> Amazon > GCM -> Android设备
确定了一个与芹菜和Rabbit
浏览 0提问于2018-04-11得票数 0
我有java Kafka流处理应用程序和python应用程序。Java应用程序生成数据，python使用者使用它。当processing.guarantee设置为exactly_once时，python使用者无法反序列化数据。反序列化失败。
我尝试了一个java使用者，而java使用者正在成功地读取数据。然后，我在java应用程序中将processing.guarantee返回到atleast_once。现在，python应用程序能够读取而不会出现任何问题。
我从控制台使用者处检查了有效负载，在exactly_once和atleast_once两种情况下，有效负载看起来都是一样的。在两种情况下
浏览 11提问于2022-08-18得票数 0
回答已采纳
在我们的系统中，9个进程同时写入相同的CSV输出。输出速度快。每天大约有一千万行新行。为了编写CSV文件，我们使用Python2.7的csv模块。
最近，我注意到CSV文件中有一些混合行(参见下面的示例)。
例如：
"name", "sex", "country", "email"
...# skip some lines
"qi", "Male", "China", "redice
...# skip some lines
"Jamp", "Male
浏览 3提问于2013-11-07得票数 5
回答已采纳
我正在发布相同的数据(主题，键和值)从基于python confluent_kafka库的生产者和基于Java apache库的生产者，但当消息检查卡夫卡，然后他们被发布到不同的卡夫卡分区。
我期望在默认情况下，这两个库都将使用相同的散列方法(murmur2)，并在将消息发布到Kafka时确定相同的分区，但看起来这并没有发生。
是否需要在Python库上设置标志或选项，以便它使用相同的算法并生成相同的(与Java库相同的) Kafka分区，或者是否有其他python库可以实现这一点？
浏览 4提问于2020-02-29得票数 0
为了了解Python中的排队线程，我尝试了下面的Python代码(来自在线教程)，但我想知道为什么生产者和消费者函数的输出行要多得多。我在一台具有i7 cpu内核的Windows机器上运行了Python。输出在。一些进一步的问题：
主函数指示流水线中的maxsize=10，因此如果我们将输出中的消费者和生产者计算在内，应该有20行，如果max_workers=2乘以2，总共应该有40行输出，对吗？但我得到了超过300行的输出，不知道为什么。它们是在多个cpu核上运行还是在发生其他事情？
此外，使用者输出行总是显示size=1，但它们不应该显示介于1到10之间的范围，以反映管道大小吗？
浏览 0提问于2019-08-16得票数 2
回答已采纳
假设我有三台服务器A，B & C，服务器C将有芹菜任务代码，我需要从服务器A&B执行它们。
从芹菜中，我看到有一个作为芹菜工人运行的task.py文件
from celery import Celery
app = Celery('tasks', broker='pyamqp://guest@localhost//')
@app.task
def add(x, y):
return x + y
然后我们有另一个python文件(比方说client.py)，它调用这些任务。
from tasks import add
add.dela
浏览 1提问于2020-04-22得票数 5
回答已采纳
1回答
在python程序中，我需要基于线程模块的2个线程。Thread#1生成一些数据并将其放入缓冲区中，而thread#2则负责处理缓冲区中的数据。
所以我的伪代码是这样的:线程1：
Thread#1
while True:
Generate_Some_Data()
while flag==1:
pass()
Buffer_Address=Write_It_To_Buffer()
flag=1
Thread#2
while True:
while flag==0:
pass()
Process_Data(Buffer_Address)
浏览 2提问于2012-03-07得票数 1
1回答
我在使用Python消费者时遇到了问题。
下面是我用来学习如何使用python的示例代码，它可以工作。
Producer.py
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
value_serializer=lambda x:
dumps(x).encode('utf-8'))
for e in range(1000):
data = {'number' :
浏览 3提问于2021-05-01得票数 1
我创建了一个示例程序来理解python中的threading.Condition。其中producer类将生成一些数字并发送给consumer类。consumer类将简单地取所接收数字的平方。
代码：
class ConditionProducer(threading.Thread):
""" Create numbers till limt """
def __init__(self, condition, limit, buf):
super(ConditionProducer, self).__init__
浏览 0提问于2016-02-19得票数 2
1回答
卡夫卡生产商正在发送.gz文件，但无法解压缩和读取消费者端的文件。获取错误为"IOError:而不是gzipped文件“
生产者-bin/kafka-控制台-Producer.sh-broker-list localhost:9092 -主题机场<~/下载/Stocks.json.gz
消费者-
import sys
import gzip
import StringIO
from kafka import KafkaConsumer
consumer = KafkaConsumer(KAFKA_TOPIC, bootstrap_servers=KAFKA_BROKERS)
浏览 0提问于2018-05-08得票数 2
例如，在调用ProcessPoolExecutor命令之后，主Python进程从映射获得的结果是什么结构。
浏览 0提问于2018-11-17得票数 1
我试图通过Machine_2中的python脚本向Machine_1中的Kafka主题发送一些消息。Machine_2和Machine_1都在同一个网络中，在Azure中都是VM的。
代码: sampl.py
from kafka import KafkaProducer
Producer = KafkaProducer(bootstrap_servers=['Machine_2:9092'])
Producer.send('test', 'hello')
如果我将上述代码作为
python sampl.py
没有到达Machine_2的
浏览 0提问于2017-10-13得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
