- iOS/Objective-C 元类和类别
- objective-c - -1001 错误,当 NSURLSession 通过 httpproxy 和/etc/hosts
- java - 使用网络类获取 url 地址
- ios - 推送通知中不播放声音
当我运行 Kafka Streams 应用程序的多个实例时,只有第一个实例正确接收消息。但是,如果我启动新实例,它们将收不到任何消息。
有什么解决这个问题的建议吗?
这是我的 Kafka 流媒体应用
package test.kafkastream;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TopologyBuilder;
public class Main {
public static void main(String[] args) {
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount-processor");
//props.put(ConsumerConfig.GROUP_ID_CONFIG, "streams-wordcount-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.38:45983,192.168.2.112:45635,192.168.2.116:39571");
//props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
//props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);
// setting offset reset to earliest so that we can re-run the demo code
// with the same pre-loaded data
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
TopologyBuilder builder = new TopologyBuilder();
builder.addSource("Source", "topic6");
builder.addProcessor("Process", new ProcessMessage(), "Source");
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();
}
}
这是我的制作人
package test.kafkamesos;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
public class Producer {
public static void main(String[] args) throws InterruptedException, ExecutionException {
Map<String, Object> producerConfig = new HashMap<String, Object>();
producerConfig.put("bootstrap.servers", "192.168.2.38:45983,192.168.2.112:45635,192.168.2.116:39571");
//producerConfig.put("bootstrap.servers", "localhost:9092");
// optional:
producerConfig.put("metadata.fetch.timeout.ms", "3000");
producerConfig.put("request.timeout.ms", "3000");
// ... other options:
// http://kafka.apache.org/documentation.html#producerconfigs
ByteArraySerializer serializer = new ByteArraySerializer();
KafkaProducer<byte[], byte[]> kafkaProducer = new KafkaProducer<byte[], byte[]>(producerConfig, serializer,
serializer);
int i = 0;
while (true) {
String message = "{data:success,g:" + i + "}";
ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>("topic6", message.getBytes());
kafkaProducer.send(record).get();
System.out.println("sending " + message);
Thread.sleep(1000);
i++;
}
}
}
和我的 Dockerfile
FROM openjdk:8-jre
COPY ./target/*-with-dependencies.jar /jars/service-jar.jar
CMD java -cp /jars/service-jar.jar test.kafkastream.Main
最佳答案
我认为您遇到此问题是因为 Kafka 代理仅针对您正在使用的主题 (topic6
) 配置了一个分区。来自 Confluent 博客:
For example, if your application reads from a single topic that has 10 partitions, then you can run up to 10 instances of your applications (note that you can run further instances but these will be idle). In summary, the number of topic partitions is the upper limit for the parallelism of your Streams API application and thus for the number of running instances of your application.
来源:https://www.confluent.io/blog/elastic-scaling-in-kafka-streams/
关于java - Kafka Streaming 不适用于多个实例,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43440611/
有没有一种方法可以使用标准类型构造函数(例如 int、set、dict、list、tuple 等)以用户定义的方式将用户定义类的实例强制转换为其中一种类型?例如 class Example:
我知道这个问题在Stackoverflow中有很多问题,但是即使有很多答案,这些答案也帮不了我什么,也没有找到答案。 在我的WebAPP中,它可以正常工作,但是当我将其转换为API时,它失败了(主题标
这个问题已经有答案了: Why does the ternary operator unexpectedly cast integers? (3 个回答) 已关闭 9 年前。 最近遇到一个Java的陷
我尝试使用 FirebaseApp.configure() 配置 Firebase,但遇到以下崩溃: *** Terminating app due to uncaught exception 'c
我有一个自连接员工实体类,其中包含与其自身相关的 id、name 和 ref 列。我想创建它的新实例并将其保存到数据库。 首先我创建了一个 Employee 类的实例并将其命名为 manager。然后
我有一个用于添加新公寓的表单,在该表单中我有一个下拉列表,用户可以在其中选择负责的人员。 显然,当您从下拉列表中选择并尝试保存公寓时,我的应用程序认为该人已被修改。它给了我下面的错误,指示我应该首先保
从 Visualforce 页面,我需要检索我们组织的 salesforce 实例的 URL,而不是 Visual Force URL。 例如我需要https://cs1.salesforce.com
我遇到了一些可能的问题答案,但这是关于从 Hibernate 3.4.0GA 升级到 Hibernate 4.1.8 的问题。所以这曾经在以前的版本下工作,我已经四处搜索了为什么它在这个新版本中出现了
似乎一遍又一遍地问这个问题,我仍然找不到解决我问题的答案。我在下面有一个域模型。每个新创建或更新的“安全用户”都需要我确保其具有配置文件,如果没有,则创建一个新的配置文件并分配给它。 配置文件的要求相
我很难调试为什么 JPA 不级联我的 @ManyToMany 关系。我发现的所有答案都与缺少级联语句有关。但我确实拥有它们并且仍然得到: Caused by: org.hibernate.Transi
Play 服务 API 表明有一个叫做 Instance ID 的东西 但是,在 Android Studio 中包含以下内容后,我无法导入 InstanceID 类 compile "com.goo
我正在使用 Seam 框架。我有 2 个实体: 请求.java @Entity @Table(name = "SRV_REQUEST") public class Request { private
This question处理构建一个适当的Monad来自单子(monad)的实例,但仅在某些约束下 - 例如Set .诀窍是将其包装成 ContT ,它将约束推迟到包装/展开其值。 现在我想对 Ap
我正在尝试执行此查询: StringBuffer sb = new StringBuffer(); sb.append("select p from PointsEntity p " + "where
我试图了解是否可以更改我的 hibernate 配置并使用单个 MySQL 实例(而不是我当前拥有的多个 MySQL 实例): 我有一个使用 hibernate 的 Java 应用程序,与 2 个模式
我有一个选项卡滑动布局,其中包括四个选项卡,每个选项卡都有自己的布局和 fragment ,在我的主要 Activity 布局中,viewpager 参与更改选项卡。特定 View (选项卡)在应用程
我看到很多帖子声称他们正在运行 MySql 的 RDS 实例,但无法连接到该实例,但我没有运行 RDS。 我使用 EC2 实例来托管我的 WordPress 博客,该博客是使用 Web 平台安装程序安
因为我在我的 ec-2 实例上的 python 虚拟环境中运行应用程序( Airflow ),并且我想在同一个 ec2 实例上的默认 python 环境中运行命令,所以我认为 ssh 到我自己的实例更
这个问题已经有答案了: How to fix the Hibernate "object references an unsaved transient instance - save the tra
例子: run APP1 .. ... run APP1 ... run APP2 如何在 APP2 中对 Vue 说我需要调用 APP1?
我是一名优秀的程序员,十分优秀!