当前位置: 首页 > 知识库问答 >
问题:

用户在Apache Kafka中未接收到消息

巫新知
2023-03-14
        Properties properties = new Properties();
        properties.put("metadata.broker.list","Running kafka ip addr:9092");
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        ProducerConfig producerConfig = new ProducerConfig(properties);
        kafka.javaapi.producer.Producer<String,String> producer = new kafka.javaapi.producer.Producer<String, String>(producerConfig);
        String filePath="filepath";
        File rootFile= new File(filePath);
        Collection<File> allFiles = FileUtils.listFiles(rootFile, CanReadFileFilter.CAN_READ, TrueFileFilter.INSTANCE);
        for(File file : allFiles) {
            StringBuilder sb = new StringBuilder();
            sb.append(file);
            KeyedMessage<String, String> message =new KeyedMessage<String, String>(TOPIC,sb.toString());
            System.out.println("sending msg from producer.."+sb.toString());
            producer.send(message);
        }
           producer.close();
         properties.put("bootstrap.servers","Running zookeaper ip addr:2181");
         properties.put("group.id","test-group");
         properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         properties.put("enable.auto.commit", "false");

         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
         consumer.subscribe(Collections.singletonList(topicName)); 
         while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                {
                    System.out.println("topic = "+record.topic());
                    System.out.println("topic = "+record.partition());
                    System.out.println("topic = "+record.offset());
                }
                try {
                  consumer.commitSync(); 
                } catch (CommitFailedException e) {
                    System.out.printf("commit failed", e) ;
                }
            }
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>0.10.0.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.10.1.0</version>
    </dependency>

我从以下链接获得所有信息:
https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/kafkaConsumer.html

当我们运行consumer时,我们没有从consumer端得到任何通知。请给我一点主意。

共有1个答案

郑安晏
2023-03-14

对于生产者:

   properties.put("metadata.broker.list","Running kafka ip addr:9092");

我猜,这应该是“bootstrap.servers”。

对于消费者:

properties.put("bootstrap.servers","Running zookeaper ip addr:2181");
 类似资料:
  • 概述 为了能够让轻应用订阅号的开发者接收到用户在消息窗口的留言消息,开发者可以在管理后台设置消息服务器并开启接收用户对话消息模式。 设置消息服务器时接需要提供可用的接收消息的回调URL地址,为了让通信更加安全,建议使用https。 设置成功并开启了接收对话消息模式后,用户在轻应用或订阅号窗口里发送的消息会推送给设置的URL,服务器接收到消息后,可以通过异步发送消息接口给用户回复消息。 设置消息服务

  • 消费者输出: 我对Kafka是新手,不知道我错过了什么来让它正确地工作。根据Kafka文档,消息的顺序是为我的用例保证的,所以一定有一些愚蠢的错误,我正在做,无法找出它。 除了Kafka,我还有别的选择吗? 谢谢

  • 我一直在努力通过Smack和Openfire服务器与XMPP聊天。我的问题如下: 每当一个用户向另一个用户发送消息时,该消息就会在另一个用户处正确接收。但是任何回复都不会出现在第一条消息的发件人处。因此,用户1成功地发送给用户2。然后,用户2无法向用户1发送任何回复。另一方面,如果我重新启动并让用户再次登录,则用户2可以发送给用户1,但反之亦然。 我想说的是,只有聊天的发起者才能发送消息,接收者不

  • 我正在使用Spring stomp websocket框架向客户机发送订阅消息。我们使用ActiveMQ作为消息代理,并使用stomp javascript客户机。Spring 4.1.5和在此体系结构中,消息是通过以下方式发送的: 我只看到一个客户机接收到消息,而另一个没有。通过阅读Spring文档,我了解到这是默认的行为。谁能告诉我我做错了什么。 多谢了。

  • 当监听设备后,会返回接收到的消息数据。 请求方式: 无 返回值: "|4|2|5|message|" 返回接收到的消息 参数 message 返回的消息内容

  • 我的Spring/java消费者无法访问生产者生成的消息。但是,当我从控制台/终端运行消费者时,它能够接收Spring/java生产者生成的消息。 消费者配置: 监听器配置: Kafka听众: 消费者应用: 测试用例1:通过 我启动了我的Spring/ java生产者并从控制台运行消费者。当我生成消息表单创建者时,我的控制台使用者能够访问该消息。 测试用例2:失败:我启动了spring/java消