我有一个场景，我将不得不在Nifi Kafka消费者处理器中合并消息，通过Kafka生产者Jar发送的消息在本地调用..i能够在kafka消费者控制台中消费消息，但不能在Nifi...how中查看消息，如果消息为json格式
浏览 23提问于2020-02-17得票数 0
3回答
我有一个Spring应用程序，它听了Spring的一个Kafka主题。在使用消息之后，将使用消息中的信息执行几次web/rest服务调用，以收集其他一些数据，这一过程需要一些时间才能完成。因此，我使用一个大小为20的线程池来创建并行消息处理。
这个系统通常运行良好，但很少有大量的消息(大约200 K)在短时间内(1秒)向Kafka主题发送/生成消息。在这种情况下，消费者会立即使用消息，但是消息处理机制还不够快。因此，当等待线程和应用程序获得OutOfMemoryError时，所有消耗的消息都保留在内存中。
将线程池大小增加到某个点可能是一种改进，但它不是解决此问题的永久解决方案。我希望在一段时
浏览 4提问于2021-01-02得票数 0
1回答
我目前正在从事一个项目，在这个项目中，我们可以在微服务体系结构中实现kafka。您是否成功地创建了mS到kafka主题交互的合同测试用例，请使用pact？
我的实现是microservice1向REST客户端发布一条消息，后者将消息发布到Kafka主题。microservice2使用GET方法从Kafka主题中检索消息。
浏览 4提问于2017-07-03得票数 15
回答已采纳
1回答
在我发布我的问题之前，我想感谢Gary和Artem帮助我解决我的问题，因为我能够成功地发布从JMS到Kafka的消息，并且事务已经就绪。
现在，我正面临着另一个问题，并测试当我的卡夫卡下降时会发生什么。当kafka关闭前几次重试时，kafka出站适配器抛出异常，消息返回到JMS，然后一次又一次地重试。
然而，经过几次重试，即使kafka关机，消息也会从JMS中退出队列，我得到以下例外情况：
2017-07-10 23:27:51.117 ERROR 16116 --- [enerContainer-1] o.s.k.support.LoggingProducerListener : Ex
浏览 2提问于2017-07-10得票数 0
回答已采纳
要从SQL Server触发器内部生成kafka消息，必须使用哪些选项？
即，当我更新一行时，我想将消息发布到kafka集群。
我想到的唯一选择是实现一个REST代理，并从触发器过程中调用它。还有其他方法吗？
浏览 2提问于2015-08-28得票数 0
我想做什么
我有一个用Flask编写的HTTP服务，它是一个用于构建不同服务实例的模板。因此，这个模板需要是可泛化的，以处理实际使用和不包含Kafka消费的用例。
我的目标是让一个可选的Kafka消费者在API模板的背景下运行。我希望任何需要它的服务都能够异步地从Kafka主题读取数据，同时也像通常那样独立地响应HTTP请求。这两个进程(Kafka消费，HTTP请求处理)并不相关，只是它们将发生在同一服务的幕后。
我写的东西
这是我的装置：
# ./create_app.py
from flask_socketio import SocketIO
socketio = None
def
浏览 4提问于2019-10-23得票数 0
1回答
Camel JMS组件多线程
、、、、
我有一个spring boot应用程序，使用Camel的jmsComponent处理来自多个Solace队列的消息。我为每个队列连接定义了一个Camel route，并定义了一个将消息转换为json的Camel Processor。有大量来自队列的消息，而我的程序无法跟上速度。 路由： @Component
public class SomeRoute extends RouteBuilder {
@Override
public void configure() throws Exception {
from("jms:{{queue.name}}&
浏览 22提问于2020-07-21得票数 1
我正在尝试在Spark Structured和Hazelcast Jet之间映射一些概念，并理解其他主题。
Q1 -在Spark中，每个Kafka分区将成为spark中的一个分区，然后它们将由单个任务并行处理。我想我在某处读到过Hazelcast Jet将合并来自kafka的所有消息，而不考虑group.id和主题分区，对吗？
Q2 -我们如何在Jet程序中增加“消费者”的数量，以增加来自kafka的吞吐量消耗？在Spark中，我想我们只需要增加主题分区的数量，以便为新分区分配新的spark任务。
Q3 -如果上面的Q1是真的，有可能避免合并和分发要并行处理的kafka分区吗？一旦消息已经在k
浏览 0提问于2018-01-23得票数 2
我第一次在kafka中使用Node，使用的是kafka-node。消费消息需要调用外部API，这甚至可能需要一秒钟的响应时间。我希望克服我的消费者的突然故障，如果一个消费者发生故障，将取代它的另一个消费者将收到相同的消息，即其工作尚未完成。
我正在使用kafka 0.10，并尝试使用ConsumerGroup。
我考虑在options中设置autoCommit: false，并仅在消息的工作完成后才提交消息(就像我之前对一些Java代码所做的那样)。
但是，我似乎不能确定只有在消息完成后才能正确地提交消息。我应该如何提交它？
我的另一个担心是，由于回调，下一条消息似乎在前一条消息完成之前就被读
浏览 0提问于2016-10-18得票数 4
我正在尝试处理失败，并将消息写入failure-Q，以防消息在db期间由于db停机或任何其他问题而失败。
我们需要每天重试一次failure-Q的消息。因为我们使用属性文件来配置消费者配置：
# KAFKA
spring.kafka.bootstrap-servers=localhost:9092
## CONSUMER
kafka.topic.consumer.boot: tink_boot.j
kafka.consumer.group-id: tink_boot.j-001
### Kafka from 5000 to 60000
spring.kafka.consumer.auto-c
浏览 6提问于2020-03-27得票数 0
1回答
我目前正在尝试编写一个适配器，它将使用来自ActiveMQ的消息并将其发布到Kafka。
我正在考虑使用spring集成来集成这两个消息传递系统。
我的问题是，我的应用程序不会维护模型的注册表，许多应用程序将使用这些模型将记录发布到activeMQ。我想要接收这些javax消息，并想执行一些转换，比如将jmscorrelationId添加到kafka消息中。
ALso，另一个要求是仅当kafka发送/发布成功时才向活动mq发送确认。
是否可以使用spring集成将ack发送回activemq？
spring集成是个不错的选择吗？
请注意，我的技术架构师不赞成使用Camel/Mule。此外，他不
浏览 10提问于2017-07-05得票数 0
回答已采纳
当使用时，我可以通过设置我的kafka生产者的kafka.compression.codec属性来设置编解码器。
假设我在我的生产者中使用了snappy压缩，当我使用某个kafka消费者消费kafka的消息时，我应该做一些事情来解码来自snappy的数据还是它是kafka消费者的内置功能？
在中，我找不到任何与kafka consumer中的编码相关的属性(它只与生产者相关)。
有人能把这个清理一下吗？
浏览 0提问于2013-11-10得票数 9
回答已采纳
1回答
我需要将消息接收到我的BigQuery表，我想知道使用这些消息的最佳方式是什么。我在AWS的Kafka服务器会生成AVRO消息，根据我所看到的，Dataflow需要接收JSON格式的消息。因此，我搜索并找到了一篇文章，解释了如何向PubSub接收消息，但在PubSub上，我只看到在这种体系结构中，他们在GCP上创建了一个Kafka来生成消息。我需要知道的是：
可以从外部的Kafka服务器在PubSub上接收AVRO消息，然后使用messages反序列化消息，将其发送到Dataflow，并最终将其发送到BigQuery表？
还是我需要创建一个Kafka并使用它来使用来自外部服务器
浏览 1提问于2021-04-27得票数 3
回答已采纳
此接口- rd_kafka_poll_set_consumer()的用途是什么？在rdkafka.h文件中，它指出一旦您将rd_kafka_poll队列(rk->rep)重定向到rd_kafka_consumer_poll的队列(rkcg->rkcg_q)，在此之后您就不应该调用rd_kafka_poll。
谁能详细解释一下哪个队列会发生什么？producer是否应该调用rd_kafka_consumer_poll来查看生成的消息？
浏览 0提问于2016-10-01得票数 0
作为我的客户端站点设计决策的一部分，http请求-响应流中涉及的组件(微服务)允许生成关于kafka主题()的消息，但不允许使用来自kafka主题的消息。
这类组件(微服务)可以读写数据库，可以与其他组件对话，可以生成主题消息，却不能使用来自卡夫卡主题的消息。
相反，设计建议编写单独的实用程序，使用kafka主题的消息并将其存储在数据库中。涉及请求-响应流程的组件(微服务)将从数据库读取该信息.
如果这些组件(微服务)消耗了kafka的主题，那么设计缺陷是什么呢？为什么设计建议编写单独的实用程序来使用kafka主题并存储在数据库中，以便组件可以从数据库中读取这些信息。
浏览 2提问于2021-01-29得票数 0
这可能是数据丢失的情况吗？如果由于不可纠正硬件问题，kafka的请求队列排队，如果这一次，我们关闭/退回kafka代理，那么追随者将会发生什么？队列中的消息会发生什么？ kafka.network:type=RequestChannel,name=RequestQueueSize
Size of the request queue. A congested request queue will not be able to process incoming or outgoing requests 根据我从kafka那里学到的，这应该在networklayer中，这是否意味着队列中的消息将被
浏览 45提问于2021-08-25得票数 0
2回答
在我们公司的最后一个项目中:客户端发出身份验证请求，然后我们的应用程序的第一层获取客户请求并在Kafka上生成消息，我们的核心服务消费该消息，然后向银行服务发出rest请求，在得到响应后，我们在Kafka上生成响应消息，然后我们的应用程序的第一层向客户发送消息。它是真正的Kafka用例还是删除第一层和Kafka，在客户端和核心之间使用rest服务更好。谢谢
浏览 0提问于2021-02-21得票数 1
回答已采纳
2回答
我创建了微服务发送多个通知，如电子邮件，短信等，这是消费kafka的通知。我希望像免打扰这样的用户首选项被视为用户不应该在指定的免打扰期限内收到通知。
我如何才能获得免打扰功能，因为kafka会不断地向我发送带有时间戳(发送消息的日期和时间)的消息，并确认用户在订阅的免打扰期间不会收到这些通知？
我能得到一些帮助吗？
浏览 0提问于2020-08-27得票数 0
1回答
我们计划在Java中有一个Kafka用户作为守护进程，它需要嗅探Kafka的消息，但是我们想知道我们如何能够自动调节这个守护进程。请建议一个最好的设计。
通过这个AWS员工，这是一个好的解决方案吗？
浏览 4提问于2016-05-22得票数 0
回答已采纳
我们正在尝试连接AS400中的一个进程，以便将信息传递到SQL Server2016中的数据库。我们有一台Apache Kafka服务器正在运行并准备就绪。使用Apache Kafka将i系列连接到SQL Server的最佳方式是什么？以这样的方式，RPGLE程序可以传递订阅了Kafka中特定主题的消息。
我们应该安装一些额外的东西吗？有没有开源的选择？
我们应该配置什么？
浏览 0提问于2021-04-24得票数 0
Java应用程序向Kafka集群发送或发布消息。发送记录的代码如下所示。由于这个应用程序向kafka发送了大量的记录消息，因此调试起来有点困难。但我看到一些消息没有发送给kafka，因为我在kafka broker subscriber中没有看到相同主题的一些消息。所以看起来这条管道中有一些数据泄漏。 为此，我添加了java.util.concurrent.Future.isDone()方法来查看响应。它返回的结果是false。所以我很困惑到底哪里存在泄漏。是否无法向kafka发送记录？或者在Kafka broker中，在将记录放入主题之前，它无法处理记录？ // setting prope
浏览 29提问于2019-09-20得票数 0
1回答
我有一个集成流程绑定到Kafka的云流输入通道。
然后使用singleThreadExecutor将消息发送到publishSubscribe通道。
从那里，它们转到一个处理它们的订阅者。处理可能需要一些时间。
据我所知，处理是在singleThreadExecutor上完成的。而云流媒体的线程被释放，以便从Kafka获取另一条消息。
如果新消息到达，但处理线程仍然繁忙，会发生什么情况？云流的线程会等待还是消息会被丢弃？如果它会等多久呢？是否有一些默认的超时值？
我有一个猜测，我们在这种情况下丢失了一些消息，因为我可以在Kafka中看到消息，但在DB中没有相应的更新……
但大多数消息都按预期进
浏览 0提问于2020-07-23得票数 0
如果有rd_kafka_flush接口，rd_kafka_outq_len的作用是什么？考虑到使用rd_kafka_outq_len只是为了确保所有生成的消息都到达broker？难道我们不能直接调用rd_kafka_flush而不使用rd_kafka_ouq_len吗？(它现在过时了吗？)
浏览 13提问于2016-10-20得票数 0
我如何测试/产生一个飞机例外代码14？
我有一个简单的带有Aerospike的节点环境和一个K8s上的java应用程序。
应用程序有3个荚，它们都是使用Kafka主题的消息，有3个分区，都在同一个消费者组中。
使用Kafka生产者驱动程序，我们立即注入200条消息，没有Kafka消息键(这样kafka将在3个主题分区上循环运行)。
所有消息都与相同的Aerospike键相关，因此3个应用程序荚应该并行更新相同的记录，从而导致Aerospike热键异常(KEY_BUSY，错误代码14)。但这并没有发生，所有200条消息都被成功地处理了。
在aerospike.conf中，配置参数“事务-挂起-限
浏览 2提问于2022-03-10得票数 1
1回答
这就是我的消费者的初始化方式： const client = new kafka.Client(config.ZK_HOST)
const consumer = new kafka.Consumer(client, [{ topic: config.KAFKA_TOPIC, offset: 0}],
{
autoCommit: false
}) 现在是消费者consumer.on('message', message => applyMessage(message)) 问题是，applyMessage使用knex与数据库通信，代码看起来像这样： async fu
浏览 5提问于2019-01-18得票数 1
回答已采纳
获得了一个现成的Kafka服务器，并有以下脚本
#!/usr/bin/perl
use Net::Kafka::Producer;
use AnyEvent;
my $condvar = AnyEvent->condvar;
my $producer = Net::Kafka::Producer->new(
'bootstrap.servers' => 'localhost:9092'
);
for (my $index = 1;;$index++) {
my $msg = "message:
浏览 15提问于2022-03-20得票数 1
回答已采纳
16回答
我被要求评估RabbitMQ而不是Kafka，但我发现很难找到一个消息队列比Kafka更合适的情况。有人知道消息队列在吞吐量、持久性、延迟或易用性方面更适合的用例吗？
浏览 18提问于2017-02-10得票数 509
回答已采纳
我有一个叫spring-cloud-starter-parent:Hoxton.SR9的项目。这里的主要目标是消息传递，因此有消息(kafka/rabbit)生产者和消费者声明： spring:
cloud:
stream:
default-binder: kafka
contentType: application/*+avro
bindings:
loanContractWaitingForActivationInput:
binder: rabbit
destination: ice
浏览 38提问于2021-09-28得票数 0
回答已采纳
1回答
我有一个java服务器端应用程序，它侦听tcp连接，读取消息，处理和发布到kafka主题。我们使用带有kafka的事务性kafka模板来实现一次，只有当我们通过semantics.This进行记录时，record.We才能很好地提交一批记录(检查消息是否具有某个参数集，然后提交)或基于时间间隔。这是卡夫卡template.We使用SpringKafka2.1的可能性吗？
浏览 0提问于2018-09-14得票数 1
回答已采纳
有些地方提到的Kafka是发布-订阅消息传递。其他消息来源提到的Kafka是消息队列。我可以问一下它们之间的区别吗? Kakfa可以用作数据库吗？
浏览 54提问于2019-11-18得票数 0
1回答
我是新的微服务架构，并希望创建一个集中的通知微型服务，以发送电子邮件/短信给用户。
我的第一个选择是创建一个通知Kafka队列，所有其他微服务都可以将通知发送到该队列。然后，通知微服务将侦听此队列并相应地发送消息。如果重新启动或删除通知服务，则不会丢失任何消息，因为消息将存储在队列中。
我的第二个选择是在通知微服务上添加一个通知消息API。这将使所有其他微服务变得更容易，因为它们只需要调用API，而不是与队列集成。然后API将在内部将消息发送到通知Kafka队列并发送消息。这里唯一的问题是，如果API不可用或出现错误，我们将丢失消息。
对处理这件事的最佳方法有什么建议吗？
浏览 1提问于2019-02-28得票数 3
回答已采纳
1回答
我已经安装了生产和消费信息的kafka客户端，当我们将有效载荷从生产者发送到另一个主题时，它可以正常工作，所以我现在遇到了第一条消息的问题--我可以把它发送到主题，我也可以从kafka主题中使用它，现在我可以发送第二条消息了，但是消费者并没有读取来自kafka主题的第二条消息，知道这里发生了什么吗？
producer.js
var config = require('./config.js');
var zk = require('node-zookeeper-client');
var kafkaConn = "130.8";
var kafk
浏览 0提问于2016-06-24得票数 0
回答已采纳
1回答
卡夫卡消费者在数据包中规定时间后，消费者是否可以选择数据包，或者我们如何在kafka中实现这一目标？
找到了相关的问题，但没什么用。正如我所看到的: Kafka基于文件系统的顺序读取，只能用于读取主题，直接保持消息的顺序。我说的对吗？
rabbitMQ也有同样的可能。
浏览 1提问于2022-06-09得票数 1
1回答
我有以下场景。 我们在K8s的云中对我们的应用程序进行了负载测试。我们的入站消息是通过Kafka发送的，我们正在给Kafka回信。我们的架构是这样的: Kafka线程将消息推送到disruptor (阻塞等待策略，512大小)，业务线程将消息从disruptor中取出进行处理。为了模拟负载，我们在Kafka主题(4个分区)中设置了应用程序未运行时的关闭500K消息。然后，我们启动应用程序来测量负载。 我们看到的是，disruptor完全填满了剩余的0容量，然后开始耗尽，这种情况一次又一次地继续下去。 这是正确的行为，还是我们以错误的方式使用颠覆者？请分享你的想法？
浏览 35提问于2020-07-21得票数 0
我有一个类似队列的列族，大约每小时更新一次。几个小时或一天后，cassandra有很多阅读暂停。我使用Cassandra得到这个错误消息：
code=1100 [Coordinator node timed out waiting for replica nodes'
responses] message="Operation timed out - received only 0 responses."
info={'received_responses': 0, 'required_responses': 1, 'cons
浏览 0提问于2016-05-16得票数 0
我有一个风暴拓扑来处理来自Kafka的消息，并根据手头的任务在Cassandra中进行HTTP调用/保存。我一收到消息就会处理。然而，由于外部源(如HTTP )的响应，很少消息没有被完全处理。我想实现一种指数退避机制，以便在一段时间后重新尝试HTTP服务器不响应/返回要重试的错误消息。我想不出几个能实现它们的想法。我想知道其中哪一种解决方案会更好，如果有任何其他的解决方案，我可以使用，这是容错。因为这用于实现指数退避，所以每条消息都有不同的延迟时间。
在Kafka中发送另一个主题，稍后再使用。我喜欢的解决方案。我知道我们可以使用卡夫卡偏移，所以在后一阶段消费信息。我怎么也找不到文档/示例代
浏览 0提问于2016-02-13得票数 5
回答已采纳
2回答
卡夫卡车轮定时器
、、、
日安，
我想知道kafka队列是否可以保存几秒钟的数据，而不是发布数据。
我收到一条来自kafka主题的消息，在解析完数据后，我将它保存在内存中一段时间(10秒)(这是随着唯一消息的传递而建立的)，每条消息都有自己的计时器)，我希望kafka告诉我消息过期了(10秒)，这样我就可以继续执行其他任务了。
但是，由于flink/kafka是事件驱动的，我希望kafka有某种圆形的计时轮，它可以在10秒后复制给消费者的信息的密钥。
对如何使用flink窗口或卡夫卡功能来存档这一点，我有什么想法吗？
问候
浏览 0提问于2019-02-19得票数 1
我正在创建两个apache camel (blueprint XML) kafka项目，一个是kafka-producer，它接受请求并将其存储在kafka服务器中，另一个是kafka-consumer，它从kafka服务器获取消息并处理它们。 这种设置对于单一主题和单一使用者来说工作得很好。但是，如何在同一个kafka topic中创建独立的消费群？如何在不同消费者组内的同一主题内路由多个特定于消费者的消息？如有任何帮助，我们将不胜感激。谢谢你。
浏览 28提问于2020-05-03得票数 1
5回答
Kafka根据生产者分配的分区将传入的消息分割成分区。来自分区的消息随后由不同消费者组的消费者使用。
这种架构让我对使用Kafka作为工作/任务队列感到谨慎，因为我必须在生产时指定分区，这间接地限制了消费者可以在其上工作，因为一个分区只发送给一个消费者组中的一个消费者。我不愿提前指定分区，以便任何可用的使用者都可以完成该任务。是否有一种方法可以在Kafka体系结构中构造分区/生产者，在这种结构中，任务可以由下一个可用的使用者来拉，而不必在工作产生时选择一个分区来提前拆分工作？
仅为该主题使用一个分区会将所有任务放在同一个队列中，但随后每个使用者组的使用者数量限制在1，因此每个使用者必须在不同的
浏览 8提问于2016-03-24得票数 70
回答已采纳
1回答
我正在使用多个使用Kafka进行异步通信的应用程序。这些应用程序由多个部门管理，合同测试适用于确保通信期间使用的消息遵循预期的模式，并将根据合同规范进行演变。
听起来python的pact库非常合适，因为它有助于为HTTP和消息集成创建契约测试。
我想要做的是发送一个HTTP请求，然后立即收听适当的、专门的Kafka主题。但是似乎测试迫使我指定HTTP代码，即使我期望的是没有HTTP状态代码的队列中的消息。此外，HTTP请求似乎是在使用者监听之前发送的。下面是一些示例代码。
from pact.consumer import Consumer as p_Consumer
from pact.p
浏览 5提问于2021-11-26得票数 0
我们目前有一个用Python编写的REST服务(让我们称之为A)，它也为用Python编写的Worker (W)触发了各种繁重的计算任务。这两个服务都是使用自写PostgreSQL队列库互连的。
他们现在就是这样交流的：
1. Call POST request to service A to create a Job in PG Queue
2. Worker takes a job from Queue and runs calculations
3. After a successful calculations the worker saves results in the sh
浏览 0提问于2022-06-15得票数 2
回答已采纳
我正在发送有关主题的消息，并且可以使用kafka控制台和Java消费者代码使用main方法读取相同的消息。但是消息并没有从使用者配置xml中读取。请帮助解决这个问题。我正在使用消息int-kafka:消息驱动-通道适配器。春季-卡夫卡1.3.0版
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://ww
浏览 2提问于2015-12-26得票数 1
回答已采纳
合同测试似乎不会为消息队列带来任何好处。有什么好处吗？
在提供者端，我们实例化一个通常进入kafka的对象。
所以提供者实际上所做的是：
数据对象实例(内存中)
序列化(由kafka图书馆提供)
序列化并用于契约合同的验证。
在消费者端，我们有相反的情况，即我们在内存中重新创建该对象并触发使用它的代码。
换句话说，消费者实际上所做的是：
从kafka队列中读取数据包(在本例中，本例中是契约的模拟队列)
将对象反序列化为对象实例
尝试触发消耗对象的代码。
在我的例子中，在使用者端，不能触发消耗对象的代码。这是因为它需要启动服务器本身的一个实例，这会带来很多
浏览 4提问于2021-05-03得票数 1
回答已采纳
让我们来讨论一种情况。
消费者收到一个Kafka流，流中的消息需要基于一些Validator Micro服务和Enricher API调用进行丰富和验证。它还将消息持久化到数据库中。
流程是这样的- Kafka Stream consumer ->每个消息->调用Rule Validator Rest API ->调用富API ->通过持久化REST API保存到数据库。
因为所有的REST调用都是同步的，并且在调用最终的Persistence Micro服务API之前，rest调用是顺序的。
显然，Kafka Stream消费者获得了高吞吐量的消息。
如何通过编码实现
浏览 0提问于2020-03-22得票数 1
我有一个卡夫卡主题，我想使用一个节点应用程序。节点应用程序必须依次处理来自主题的消息，而不是同时处理多条消息。
我尝试过这样的代码，但这不是我想做的。当主题中有消息等待处理并启动此代码时，所有消息都会立即触发on 'message‘事件。第一条消息首先获得互斥锁，但其余消息按随机顺序处理。
var mutex = require( 'node-mutex' )();
var crypto = require('crypto');
var mutexToken = crypto.randomBytes(64).toString('hex')
浏览 1提问于2017-04-19得票数 1
我遇到一个问题，一个kafka日志存储管道消耗了太多的cpu (启动时消耗了大约300%，几秒钟后消耗了100% )，但是它基本上是有效的:这个管道可以将kafka中的事件传递到elasticsearch中，没有错误消息。
logstash运行在一个docker容器中，其中包含最新版本的lostash (2.1.1，从中提取)。
docker run --rm --link kafka:kafka --link elasticsearch:elasticsearch -v "$PWD":/config-dir logstash logstash -f /config-dir/l
浏览 2提问于2015-12-28得票数 3
回答已采纳
1回答
我对Golang和Kafa并不熟悉，所以这似乎是个愚蠢的问题。
当我的卡夫卡消费者第一次连接到卡夫卡服务器之后，为什么连接到Kafka服务器和接收到第一条消息之间会有延迟(~ 20秒)？
它在consumer.Messages()之前打印一条消息，并为收到的每条消息打印另一条消息。~20秒延迟在第一fmt.Println和第二fmt.Println之间。
package main
import (
"fmt"
"github.com/Shopify/sarama"
cluster "github.com/bsm/sarama-
浏览 0提问于2018-12-07得票数 0
我正在使用，一次只获取一条消息。我只写一个话题。我可以通过一个带有文本框的pop UI屏幕来发送和接收消息。我在文本框中输入一个字符串，然后单击“send”。我想发多少信息都可以。假设我发送了3条消息，我的3条消息是“嗨”、"lol“、”拜拜“。还有一个“接收”按钮。现在，使用在中找到的传统代码，当我单击“接收”按钮时，我会同时在控制台上得到所有3条消息(嗨、lol、bye)。但是，当我单击UI上的“接收”时，我只希望只打印一条消息。例如，当我第一次按下“接收”按钮时，它会打印“嗨”，第二次是"lol“，第三次是”拜拜“。我是卡夫卡的新手，我对如何做到这一点感到困惑。我试着从代
浏览 0提问于2020-10-13得票数 2
回答已采纳
我有一个web应用程序，通过读取请求中的一些值发送kafka消息。如果我的kafka服务器宕机了，我该如何发送错误状态。现在，在服务器宕机的情况下，prodcuer不断地尝试连接，并无限地记录以下错误。
错误：
14:56:10.181 [kafka-producer-network-thread | producer-1] WARN o.a.kafka.common.network.Selector - Error in I/O with localhost/127.0.0.1
java.net.ConnectException: Connection refused
生产者配置为：
&l
浏览 0提问于2016-03-13得票数 0
在MassTransit中，当抛出异常时使用像RabbitMQ这样的传输时，消息会进入queue_name_error队列。但是使用Kafka时，没有带有_error后缀的主题，也没有支持传输的类似队列。如何使用MassTransit正确地处理异常，在哪里可以找到错误的消息？
浏览 3提问于2020-12-31得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
