Kafka多消费者
是指在Kafka消息队列系统中，允许多个消费者同时订阅和消费同一个主题的消息。这种模式可以实现消息的并行处理，提高系统的吞吐量和可伸缩性。
Kafka多消费者的优势包括：
- 高吞吐量：Kafka采用了分布式架构和批量处理机制，可以实现每秒处理数百万条消息的高吞吐量。
- 可伸缩性：通过增加消费者实例，可以实现水平扩展，提高系统的处理能力。
- 容错性：Kafka采用了分布式副本机制，可以保证消息的可靠性和容错性，即使某个消费者实例发生故障，其他消费者仍然可以继续消费消息。
- 实时处理：Kafka支持实时数据流处理，可以将消息实时传输到各个消费者，满足实时数据处理的需求。
Kafka多消费者适用于以下场景：
- 大规模数据处理：Kafka的高吞吐量和可伸缩性使其非常适合处理大规模的数据流，如日志收集、实时监控等。
- 分布式系统集成：Kafka可以作为分布式系统之间的消息中间件，实现不同系统之间的解耦和数据传输。
- 实时流处理：Kafka可以与流处理框架（如Apache Flink、Spark Streaming等）结合使用，实现实时数据的处理和分析。
腾讯云提供了一系列与Kafka相关的产品和服务，包括：
- 云原生消息队列 CKafka：腾讯云提供的高可用、高可靠的消息队列服务，支持Kafka协议，适用于大规模数据流处理和分布式系统集成。 产品链接：https://cloud.tencent.com/product/ckafka
总结：Kafka多消费者是一种允许多个消费者同时订阅和消费同一个主题的消息队列模式，具有高吞吐量、可伸缩性、容错性和实时处理等优势。在大规模数据处理、分布式系统集成和实时流处理等场景下具有广泛应用。腾讯云提供了与Kafka相关的云原生消息队列 CKafka服务，满足用户的需求。
相关·内容
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
2回答
Kafka分区延迟增加
、、、
我有一个使用Kafka 1.0作为队列的应用程序。Kafka主题有80个分区和80个消费者在运行。(Kafka-python消费者)。
通过运行以下命令：
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group mygroup --describe
我看到其中一个分区卡在偏移量上，随着新记录的添加，延迟不断增加。
上述命令的输出如下所示：
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
浏览 0提问于2017-11-15得票数 2
我用的是卡夫卡和flink。在一个简单的程序中，我使用flink FlinkKafkaConsumer09，将组id分配给它。
根据Kafka的行为，当我使用相同的group.Id在同一主题上运行两个消费者时，它应该像消息队列一样工作。我认为这应该是这样的:如果向Kafka发送2条消息，那么flink程序中的每条或一条将处理这2条消息完全两次(假设总共有2行输出)。
但是实际的结果是，每个程序将接收2条消息。
我尝试使用随kafka服务器下载的消费者客户端。它以记录的方式工作(处理了2条消息)。
我试图使用两个卡夫卡消费者在相同的主要功能的flink程序。共处理4条信息。
我还试着运行两个fl
浏览 10提问于2016-07-28得票数 6
回答已采纳
我有一个独立的Kafka设置与单盘。计划流式传输超过百万条记录。如何为我的主题确定分区以获得更好的吞吐量？必须是1个分区？
在独立Kafka服务器上，是否建议一个主题有多个分区？
浏览 2提问于2021-03-30得票数 0
这就是场景:我知道，使用与Spring相关的最新API (比如kafka 2.10)，我们可以这样做：
@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic", partitions = { "0" }) })
@KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic",
浏览 0提问于2017-04-17得票数 1
我想在我的ubuntu集群上安装Kafka的最新版本，该集群包含1个主节点和4个数据节点。
以下是我的问题：
Should kafka be installed on all the machines or only on NameNode machine?
What about zookeeper? Should it be installed on all the machines or only
on NameNode machine?
请共享在Hadoop 5节点集群中安装kafka和所需的文档。
浏览 3提问于2015-12-04得票数 2
在spring引导应用程序中，我正在设置一个kafka侦听器，而且我似乎无法使用执行器在池中运行侦听器。这是我的kafka配置：
@Bean
ThreadPoolTaskExecutor messageProcessorExecutor() {
logger.info("Creating a message processor pool with {} threads", numThreads);
ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
exec.setCorePoolSiz
浏览 1提问于2017-08-03得票数 3
回答已采纳
我正试图在两个物理机器上安装一个HLF (v2)网络。机器1：
1订购者2对等点2 couchdb 2动物园管理员2 kafka
机器2：
1秩序者2对等点2 couchdb 1动物园管理员2 kafka
集装箱在这两台机器上运行。
但是，当我试图在机器1上创建通道时
peer channel create -o orderer.example.com:7050 -c $CHANNEL_NAME -f ./channel-artifacts/channel.tx --tls --cafile /opt/gopath/src/github.com/hyperledger/fabric/peer/
浏览 4提问于2020-06-09得票数 0
我提交了多个spark streaming作业，这些作业使用相同的group.id消耗同一个kafka topic，正如纯Kafka文档中所说的，拥有相同group.id的多个消费者将加入同一个消费组，并且kafka topic中分块的记录将被拆分成这些消费者。然而，正如我在我的工作中测试的那样，这两个spark streaming作业仍然使用相同的主题的所有分区(不是均匀分割的)，并且在who进程期间没有发生重新分区/异常。这里有人知道Spark是如何管理与纯kafka平台不同的kafka分区偏移的吗？也许这是因为zookeeper管理spark中的kafka偏移量，而不是kafka本身管
浏览 63提问于2021-07-30得票数 0
1回答
我有一个具有多个主题的Kafka-Broker，每个主题都有一个分区。
我有一个消费者，它可以很好地使用主题中的消息
我的问题是，我需要通过增加分区的数量来提高消息队列的吞吐量，假设我在一个主题上有四个分区，是否有一种方法可以编写四个消费者，每个消费者指向该主题上的单个分区？
import java.util.*;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStre
浏览 25提问于2016-09-23得票数 4
回答已采纳
我对卡夫卡很陌生。我有以下架构：
1) 2 servers runing application logics and database, can I write kafka producer on these servers wrapped with docker container ?
2) 1 server reserved for kafka broker and zookeeper
3) 1 sever reserved for kafka comsumer
我很困惑
1) whether I can run kafka producer, broker and consumer
浏览 6提问于2015-11-02得票数 0
回答已采纳
在我们的项目中，有主动的Kafka服务器(PR)和被动的Kafka服务器(DR)，这两个Kafka brokers在我们的项目中都配置了相同的组名、主题名称和分区。从PR切换到DR时，在DR上手动设置_consumer_offsets。
我这里的问题是，Kafka消费者是否能够无缝地从上次读取的地方消费消息？
浏览 7提问于2020-03-31得票数 2
3回答
由于Kafka有一个基于主题的发布子体系结构，如何使用Kafka处理web应用程序的一对一和组消息部分？我正在使用SpringBoot+Angular堆栈和Docker服务器。
浏览 7提问于2020-05-09得票数 6
回答已采纳
因此，在我们的一个kafka主题中，有接近100 GB的数据。我们正在运行spark-structured来获取S3格式的数据
当数据达到10 S3时，流可以正常运行，并且我们能够以S3格式获取数据。但由于有100 it的内存，在kafka中传输数据需要花费很长时间。
问: spark-streaming如何从Kafka读取数据？是否从当前偏移量中获取整个数据？或者是成批接受一定大小的产品？
浏览 0提问于2018-12-20得票数 3
max.poll.records最近在kafka的消费者配置中被更改为500，但我想知道这对消费者调查有什么影响。它只是可以获取的最大记录数的上限，还是消费者等到它获得500条记录时才能获取。
浏览 1提问于2018-11-15得票数 5
回答已采纳
我正在"topicDemo“Kafka中生成数据和消费数据，以启用分布式数据并并行处理这些数据。
但在实时场景中，目前需要监视queue(topicDemo)中的数据数量和消耗的数据(、topicDemo、)。
有卡夫卡API提供这些细节吗？
这是我正在生成数据的代码
// create instance for properties to access producer configs
Properties props = new Properties();
// props.put("serializer.class",
浏览 0提问于2016-10-26得票数 0
据我所知，kafka消费者只能使用API轮询()从代理中提取消息，但是我们知道我们也可以使用kafka高/low级别的使用者API从代理获取消息，那么kafka高/low级消费者API和API轮询()之间的关系是什么？高/low级别的使用者API调用轮询()内部？
浏览 1提问于2017-05-15得票数 0
回答已采纳
我正在尝试从最早的偏移量中读取卡夫卡主题，然后通过python脚本对某些记录进行墓碑。由于消息数量巨大(百万+)，我希望利用多处理来使脚本在使用消息时更快。下面是脚本中的一段代码：
from kafka import KafkaConsumer
def cleanup_kafka_topic(self, env):
# Declarations
consumer = KafkaConsumer(<topic_name>, group_id=<some_group>),
boots
浏览 71提问于2021-02-19得票数 0
在Kafka中，一个消费者可以订阅的主题的最大限制是什么。我在任何地方都找不到这个记录的值。
如果消费者订阅了500000个或更多的主题，性能是否会下降。
浏览 6提问于2017-09-19得票数 5
我正在运行一个多线程卡夫卡091消费者新。
我生成client.id的方式是使用“使用者在”+ "AtomicInt“+”进程PID“上运行的主机名的组合。
当我必须停止使用者并重新启动时，我遇到了问题。使用者一直在尝试处理上一次运行中没有使用的偏移量(其中大约有100次)。但这条信息一直在失败。
2016-10-21 14:22:55,293 [pool-3-thread-6] INFO o.a.k.c.c.i.AbstractCoordinator : Marking the coordinator 2147483647 dead.
2016-10-21 14:22:55,
浏览 1提问于2016-10-21得票数 0
回答已采纳
我有一个spring boot应用程序，我想将一个Kafka主题的负载分布到3个open-shift pod中。我有下面的例子，我可以在三个不同的线程上监听3个Kafka分区，这个spring boot应用程序将加载到一个openshift pod中。但我希望能够在一个pod上从一个Kafka分区监听，这样当我在open-shift上加载3个pod时，每个pod将从一个Kafka分区监听。这将允许我将应用程序扩展到N个pod上的N个分区。我不确定这是否可能，或者是否需要使用不同的方法。谢谢
public class DepAcctInqConsumerController {
p
浏览 0提问于2021-03-11得票数 1
我在GKE上有一个Strimzi集群，我也部署了kafkaExporter。Kafka主题有acls，并且定义了一个消费者组(Spark-KafkaSource-*)，它可以从主题中读取数据。
我正在运行一个火花StructuredStreaming程序，它从卡夫卡主题中读取数据。问题是-当我检查指标KafkaExporter -> kafka_consumergroup_lag时，似乎没有向消费者组显示
消费者组出现在米制: kafka_consumergroup_members ->中。
kafka_consumergroup_members{consumergroup=
浏览 4提问于2022-08-22得票数 0
回答已采纳
几个月前，我在我的Kafka链中创建了两个消费者组(每个组有3个消费者)：
第一组:从外部源获取数据，并将它们发送给my topic1第二组:从topic1消费
这没什么问题。
最近，我加入了第三组，在我的链中加入了parralelisation：
对第一组的影响:也发送到topic3 (发送到topic1之上)新的第三组(还有3名消费者)：从topic3消费
问题:新的topic3没有被实际消耗(每个主题中只有20或50条消息在许多小时后被消耗)，在第三组的三个分区的eahc上，=>有很大的“滞后”(从未消耗过)。
在kafka日志中，我可以看到许多只针对第三组的重新平衡(似乎达到了5
浏览 1提问于2020-07-13得票数 2
我在端口8081上的kafka代理上启用了JMX。当我在jConsole中查看MBean属性时，我只看到kafka.consumer的以下内容-
kafka.consumer:type=FetchRequestAndResponseMetrics,name=FetchRequestRateAndTimeMs,clientId=ReplicaFetcherThread-2-413
kafka.consumer:type=FetchRequestAndResponseMetrics,name=FetchResponseSize,clientId=ReplicaFetcherThread-0-41
浏览 3提问于2016-06-14得票数 1
我们使用.NET Kafka客户端在C#代码中消费来自一个主题的消息。然而，它似乎有点太慢了。
我想知道我们是否可以将进程并行化一点，所以我在这里检查了这个答案：
但在下面的示例中，我并没有真正看到如何使用.NET Kafka客户端实现这种分区：
var consumerBuilder = new ConsumerBuilder<Ignore, string>(GetConfig())
.SetErrorHandler((_, e) => _logger.LogError("Kafka consumer error on Revenue response. {
浏览 0提问于2019-08-12得票数 1
1回答
有一个基本的例子，它对1个消费者来说就像一个护身符。它接收消息。但是添加一个额外的消费者将被忽略。
let kafka = require('kafka-node');
let client = new kafka.Client();
let producer = new kafka.Producer(client);
let consumer1 = new kafka.Consumer(client,[ {topic: 'topic1', partition: 0}]);
let consumer2 = new kafka.Consumer(clien
浏览 4提问于2017-02-02得票数 2
回答已采纳
我在本地服务器上安装了kafka，通过在服务器上运行的一些其他应用程序，生产人员正在将消息发布到我的kafka服务器内的代理，通过zookeeper，我可以很容易地看到我的kafka的健康状况，它显示了在我的kafka服务器中创建的所有主题，主题中的偏移量等等，所以zookeeper唯一不能显示的是单个主题中的消息，所以有人推荐kafka-manager工具，我安装并运行它，它工作得很好，它显示了来自我的kafka服务器的许多信息。但它仍然不能显示由我的kafka服务器中的各个消费者发布或消费的真实消息，所以我的问题是，是否有方法/工具/代码来找出发布或消费的消息，我的意思是除了这个kafka
浏览 7提问于2016-08-19得票数 0
kafka使用者无法仅提交特定分区的偏移量。
aklsfoipafasldmaknfa asiofuasofiusaofasd
[2019-01-04 12:22:22,691] ERROR [Consumer clientId=consumer-1, groupId=console-consumer-11955] Offset commit failed on partition my-topic-2-9 at offset 0: The request timed out. (org.apache.kafka.clients.consumer.internals.ConsumerCo
浏览 3提问于2019-01-04得票数 5
1回答
我是Kafka的新手，我正在使用Kafka 1.0。
我使用pull模式阅读Kafka消息，也就是说，我会定期poll() kafka主题以获取新消息，但我没有将偏移量写回Kafka。
我会问kafka如何知道我消耗了哪些偏移量，或者Kafka记住进度的机制是什么(Kafka偏移量)
浏览 9提问于2020-05-20得票数 1
回答已采纳
我们遇到了一个与应用程序代码从Apache的0.8.2.1版本迁移到0.9.0.0版本有关的问题。
在这种情况下，我们指的是Cloudera发布的Kafka版本：
kafka_2.10-0.8.2.0-kafka-1.3.2
kafka_2.11-0.9.0-kafka-2.0.2
在读取和写入__consumer_offsets元数据主题上的偏移时，我们检测到了这个问题。特别是，我们使用一个BlockingChannel连接到Kafka，并且在进行接收()方法调用时，我们得到一个EOFException。
特别是：
java.io.EOFException
at org.apache.kaf
浏览 0提问于2017-01-18得票数 3
回答已采纳
1回答
我有一个kafka主题被划分成60个分区。生产者以默认的循环方式分发消息。到目前一切尚好。但在消费方面，滞后正在加剧。没什么好奇怪的。令我惊讶的是，这种滞后有时是如何分布的：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
a_group a_topic 24 2013567
浏览 10提问于2022-09-27得票数 0
1回答
我已经使用python.How创建了实现的Kafka生产者-消费者消息传递主题我可以对队列做同样的事情，这样消息就只会发送给一个消费者。
这是我的生产者代码
# Import KafkaProducer from Kafka library
from kafka import KafkaProducer
# Define server with port
bootstrap_servers = ['localhost:9092']
# Define topic name where the message will publish
topicName = 'Firs
浏览 3提问于2020-12-03得票数 0
2回答
我有一个关于管理多个CG的问题，创建了三个消费者组，每个CG都有自己的kafka服务、组Id和主题。
现在我收到了预期的消息，但是，我想知道是否有可能创建下一个场景：
创建三个消费者组，但只接收来自一个用户的消息，暂时暂停/暂停其他用户组，如果他的kafka服务将下降，则使用下一个消费者组的消息，与第三个用户组相同。
下面是我的代码示例：
function createConsumerGroup(topics){
const ConsumerGroup = kafka.ConsumerGroup;
//CREATE CONSUMER GROUPS FOR EVERY SER
浏览 1提问于2018-07-12得票数 2
回答已采纳
3回答
打开卡夫卡与外界沟通的港口
、、、、
我在Azure有一个VM (Linux )，上面有Hortonworks，它发布了Kafka。
Kafka服务正在运行，我能够在VM中创建生产者和消费者。
我有服务器IP，我还可以使用8080端口登录到Ambari。
当我试图从我的Java应用程序向Kafka发送一条消息时，60秒后我会得到一个TimoutEception。
为了从VM外部为Kafka通信设置正确的端口，我需要做什么？
我认为这里的问题是Kafka在监听本地IP，而不是VM (WAN)。
任何帮助都会很感激.
浏览 2提问于2016-04-04得票数 0
回答已采纳
我是卡夫卡的新手，目前正在研究它。我在golang中使用kafka-go来创建生产者和消费者。目前，我可以创建一个生产者，但我希望消费者一旦一个主题的生产者被创建，而不是每次都被创建。意味着对于每个主题，一个消费者只创建一次。此外，当需要为一个主题创建更多的消费者来平衡负载时，它就会被创建。有没有什么方法可以通过goroutines或Faktory来安排？
浏览 18提问于2019-08-27得票数 0
当使用时，我可以通过设置我的kafka生产者的kafka.compression.codec属性来设置编解码器。
假设我在我的生产者中使用了snappy压缩，当我使用某个kafka消费者消费kafka的消息时，我应该做一些事情来解码来自snappy的数据还是它是kafka消费者的内置功能？
在中，我找不到任何与kafka consumer中的编码相关的属性(它只与生产者相关)。
有人能把这个清理一下吗？
浏览 0提问于2013-11-10得票数 9
回答已采纳
1回答
有时，kafka-node消费者从偏移量0开始消费，而它的默认行为是只消费较新的消息。则它将不会切换回其默认行为。你知道如何解决这个问题吗?知道发生了什么，它的行为突然改变了吗？代码非常简单，而且不需要修改代码就可以实现。
var kafka = require("kafka-node") ;
Consumer = kafka.Consumer;
client = new kafka.KafkaClient();
consumer = new Consumer(client, [{ topic: "Topic_23", partition: 0}
浏览 5提问于2020-01-31得票数 3
在Spring Boot项目中使用Spring for Apache Kafka时，有没有办法通过application.properties中的条目配置不同的侦听器 例如，如果我在主题foo和bar上有不同的监听程序，我是否可以有如下条目： spring.kafka.producer.foo.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.bar.value-serializer=some.other.Serializer 我想我在什么地方看
浏览 30提问于2020-03-29得票数 0
因为我是Kafka的新手，所以我能够读取文件中的记录，并通过生产者将消息发送到Kafka主题，但不能通过消费者消费相同的主题。
注意:您可以从任何文本文件读取数据，我使用的是kafka_2.11-0.9.0.0版本
下面是我的代码：
package implementation;
import java.io.BufferedReader;
//import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
//im
浏览 17提问于2017-12-21得票数 1
我们使用的是之前版本的Kafka(0.8.X)，其中所有偏移量细节都存储在ZooKeeper中。现在我们转向了卡夫卡(0.10.X)的新版本，所有的主题偏移细节都存储在Kafka本身中。我们必须将所有的主题偏移细节转移到ZooKeeper，以获取生产中现有的应用程序。卡夫卡安装在Windows机器上。我们不能从窗口运行kafka-消费群。请建议如何将提交的偏移量从ZooKeeper迁移到Kafka。
浏览 0提问于2018-02-08得票数 0
我试着用卡夫卡-客户库(0.9.0.1)来测试生产者，消费者.代理(0.9.0.1)正在服务器上运行，我已经进行了测试KafkaProducer，没有问题。但是，当我测试KafkaConsumer进行轮询时，代理会发出错误消息。
2016-03-18 13:44:19,129错误关闭/172.26.132.149套接字(kafka.network.Processor) kafka.common.KafkaException:错误请求类型10在kafka.network.Processor.read(SocketServer.scala:353) at kafka.network.Proces
浏览 0提问于2016-03-18得票数 1
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用的配置和@kafkalistener。
ReceiverConfig.java
package com.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serializ
浏览 3提问于2019-12-05得票数 1
1回答
当我检查使用者延迟时，它会显示某个特定的consumer-id是从某个主题的特定主机消费中运行的。但是当我去那个主机时，没有这样的消费者在运行。
如何杀死此consumer-id，以便可以为其所属的组重置使用者偏移量。
Kafka server version: 0.11.0.1
Kafka client version(stuck): 0.10.0.2
这个consumer-id被卡在了第一位，因为它无法使用消息，因为有些消息在Kafka中有标题。
我已经尝试过以下几种方法：
从不同的主机和不同的卡夫卡版本消费，它消费的信息，但消费者id，主机不会改变。
重新启动kafka代理，它
浏览 0提问于2019-02-18得票数 3
我正在按照快速入门指南上的说明在本地运行kafka。
然后，我在config/consumer.properties中定义了我的使用者组配置，以便我的使用者可以从定义的group.id中挑选消息
运行以下命令，
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
结果是，
test-consumer-group <-- group.id defined in conf/consumer.properties
console-consumer-67807 <-- when connecting
浏览 1提问于2020-05-13得票数 5
回答已采纳
1回答
我们一直在尝试修复不想使用数据的消费者。它每4秒轮询一次，通常只收到大约100条记录。最大拉力约为18000。然后，它将原始字节写入文件。都是快速操作。
但是消费者组呼叫显示如下
[xyz@AZ254PP01 kafka]$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --new-consumer --group postproc-red2
GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER
postproc
浏览 0提问于2016-04-11得票数 0
我有一个有40个分区的Kafka主题。在Kubernetes集群中。我还有一个使用这个主题的微服务。
在批处理过程中，有时会发生这样的情况:在某个时间点上，有一些分区留下了未处理的数据，而大多数分区已经完成。使用kafka-consumer-groups.sh时，如下所示：
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST
浏览 0提问于2020-10-02得票数 0
2回答
你好，我在研究卡夫卡。我在试着理解卡夫卡的基本知识。我正在学习卡夫卡。我用码头安装了卡夫卡。目前我有一个经纪人。我使用以下命令创建了带有3个分区的主题。
kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 3 --topic topic2
之后，我创建了如下所示的制作人。
kafka-console-producer --broker-list localhost:9092 --topic topic2
>This is my producer
我在这里完全糊涂了。当
浏览 2提问于2019-11-15得票数 1
回答已采纳
1回答
下面是如何使用Kafka作为发布订阅框架的一个很好的例子：
我正在拼命寻找一个类似的例子，展示如何使用kafka作为队列框架。这意味着一旦一个客户端检索到消息，任何其他客户端都不应该访问它。
浏览 0提问于2014-11-11得票数 0
是否可以在application.yml中仅为应用程序生成的主题启用spring.cloud.stream.kafka.streams.binder.auto-create-topics，而不是自动创建应用程序使用的主题？我希望我的制作人只负责创建主题。我正在使用spring-cloud-stream和kafka-streams绑定器。 在更一般的意义上，这是创建kafka主题的正确方法吗？
浏览 30提问于2020-11-24得票数 0
回答已采纳
在上查看此文档
Zookeper在端口2181 Broker/Kafka启动，时间为9092。
启动生成器发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka // producer is producing directly to broker not zookeeper which is correct
启动使用者接收消息
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-
浏览 15提问于2017-07-11得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
