我正在与Kafka合作,并尝试通过关注此article来设置消费者群组。唯一的区别是我创建了自己的抽象类,处理器使设计更简单。来自kafka消费者的InstanceAlreadyExistsException
下面是我的抽象类:
public abstract class Consumer implements Runnable {
private final Properties consumerProps;
private final String consumerName;
public Consumer(String consumerName, Properties consumerProps) {
this.consumerName = consumerName;
this.consumerProps = consumerProps;
}
protected abstract void shutdown();
protected abstract void run(String consumerName, Properties consumerProps);
@Override
public final void run() {
run(consumerName, consumerProps);
}
}
下面是我KafkaConsumerA
延伸上面抽象类:
public class KafkaConsumerA extends Consumer {
private KafkaConsumer<byte[], DataHolder> consumer;
public KafkaConsumerA(String consumerName, Properties consumerProps) {
super(consumerName, consumerProps);
}
@Override
public void shutdown() {
consumer.wakeup();
}
@Override
protected void run(String consumerName, Properties consumerProps) {
// exception comes from below line from two of the threads and the remaining one thread works fine.
consumer = new KafkaConsumer<>(consumerProps);
List<String> topics = getTopicsBasisOnConsumerName(consumerName);
try {
consumer.subscribe(topics);
// Setup the schema config
Map<String, Object> config = new HashMap<>();
config.put("urls", "https://abc.qa.host.com");
GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config);
while (true) {
ConsumerRecords<byte[], DataHolder> records = consumer.poll(200);
for (ConsumerRecord<byte[], DataHolder> record : records) {
Map<String, Object> data = new HashMap<>();
data.put("partition", record.partition());
data.put("offset", record.offset());
data.put("value", record.value());
System.out
.println((Thread.currentThread().getId() % 3) + 1 + ": " + decoder.decode(record.value()));
}
}
} catch (WakeupException ex) {
ex.printStackTrace();
} catch (Exception ex) {
ex.printStackTrace();
} finally {
consumer.close();
}
}
}
,下面将我的处理程序类:
// looks like something is wrong in this class
public final class ConsumerHandler {
private final ExecutorService executorServiceProcess;
private final Consumer consumer;
private final List<Consumer> consumers = new ArrayList<>();
public ConsumerHandler(Consumer consumer, int poolSize) {
this.executorServiceProcess = Executors.newFixedThreadPool(poolSize);
this.consumer = consumer;
for (int i = 0; i < poolSize; i++) {
consumers.add(consumer);
executorServiceProcess.submit(consumer);
}
}
public void shutdown() {
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
for (Consumer consumer : consumers) {
consumer.shutdown();
}
executorServiceProcess.shutdown();
try {
executorServiceProcess.awaitTermination(1000, TimeUnit.MILLISECONDS);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
}
});
}
}
这里我是从主类的消费者群体中启动我的所有消费者:
public static void main(String[] args) {
ConsumerHandler handlerA =
new ConsumerHandler(new KafkaConsumerA("KafkaConsumerA", getConsumerProps()), 3);
// run KafkaConsumerB here
handlerA.shutdown();
// shutdown KafkaConsumerB here
}
所以用这个 - 我的计划是建立一个消费群体有三个消费者KafkaConsumerA
和所有三个订阅相同的主题。
错误: -
每当我跑这个,貌似只有一个消费群体在消费工作和其他两名不起作用。我从控制台上看到这个例外:
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=a97716e0-0e05-4938-8fa1-6b872cf24e34
at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437) ~[na:1.7.0_79]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898) ~[na:1.7.0_79]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966) ~[na:1.7.0_79]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900) ~[na:1.7.0_79]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324) ~[na:1.7.0_79]
at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[na:1.7.0_79]
at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58) ~[kafka-clients-0.10.0.0-SASL.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:694) [kafka-clients-0.10.0.0-SASL.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:587) [kafka-clients-0.10.0.0-SASL.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:569) [kafka-clients-0.10.0.0-SASL.jar:na]
我在这里做什么? getConsumerProps()
方法返回属性对象,其中client.id
和group.id
与该消费者组中的所有三个消费者具有相同的值。
下面是我的设计细节:
- 我
KafkaConsumerA
将有3名消费者在消费群体和每一个消费者将在topicA
工作。 - 我的
KafkaConsumerB
(与KafkaConsumerA类似)将有两个消费者在不同的消费群体中,并且这些消费者中的每一个都将在topicB
上工作。
而这两个消费者KafkaConsumerA
和KafkaConsumerB
将运行在相互独立的不同消费者组的相同盒子上。
我明白了。谢谢(你的)信息。我会试试看。另外,你认为我的设计和逻辑是正确的,我正在尝试做什么。我在我的问题的底部提到了它。或者你看到任何可以完成的设计改进? – user1950349
这是一个非常合理的设置,但我当然不知道你的域名。 –
我的想法是在同一个JVM上运行多个kafka使用者。每个卡夫卡消费者都有多个线程。例如:'KafkaConsumerA'将有三个使用者在一个使用topicA的消费群中。 'KafkaConsumerB'将有2个消费者在另一个使用'topicB'的消费群体中。这两个'KafkaConsumerA'和'KafkaConsumerB'都将在单个JVM上运行。 – user1950349