org.apache.kafka.common.serialization.stringserializer

hfwmuf9z  于 2021-06-07  发布在  Kafka
关注(0)|答案(6)|浏览(657)

我正在创建一个简单的Kafka生产者和消费者。我正在使用Kafka2.11-0.9.0.0。这是我的制片人代码。

public class KafkaProducerTest {
public static String topicName = "test-topic-2";
public static void main(String[] args) {
    // TODO Auto-generated method stub
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer",
            StringSerializer.class.getName());
    props.put("value.serializer",
            StringSerializer.class.getName());

    Producer<String, String> producer = new KafkaProducer(props);
    for (int i = 0; i < 100; i++) {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(
                topicName, Integer.toString(i), Integer.toString(i));
        System.out.println(producerRecord);
        producer.send(producerRecord);
    }

    producer.close();
}

}

启动捆绑包时，遇到以下错误：

2016-05-20 09:44:57,792 | ERROR | nsole user karaf | ShellUtil                        | 44 - org.apache.karaf.shell.core - 4.0.3 | Exception caught while executing command
org.apache.karaf.shell.support.MultiException: Error executing command on bundles:
    Error starting bundle162: Activator start error in bundle NewKafkaArtifact [162].
    at org.apache.karaf.shell.support.MultiException.throwIf(MultiException.java:61)
    at org.apache.karaf.bundle.command.BundlesCommand.doExecute(BundlesCommand.java:69)[24:org.apache.karaf.bundle.core:4.0.3]
    at org.apache.karaf.bundle.command.BundlesCommand.execute(BundlesCommand.java:54)[24:org.apache.karaf.bundle.core:4.0.3]
    at org.apache.karaf.shell.impl.action.command.ActionCommand.execute(ActionCommand.java:83)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.karaf.shell.impl.console.osgi.secured.SecuredCommand.execute(SecuredCommand.java:67)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.karaf.shell.impl.console.osgi.secured.SecuredCommand.execute(SecuredCommand.java:87)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.felix.gogo.runtime.Closure.executeCmd(Closure.java:480)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.felix.gogo.runtime.Closure.executeStatement(Closure.java:406)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.felix.gogo.runtime.Pipe.run(Pipe.java:108)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.felix.gogo.runtime.Closure.execute(Closure.java:182)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.felix.gogo.runtime.Closure.execute(Closure.java:119)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.felix.gogo.runtime.CommandSessionImpl.execute(CommandSessionImpl.java:94)[44:org.apache.karaf.shell.core:4.0.3]
    at org.apache.karaf.shell.impl.console.ConsoleSessionImpl.run(ConsoleSessionImpl.java:270)[44:org.apache.karaf.shell.core:4.0.3]
    at java.lang.Thread.run(Thread.java:745)[:1.8.0_66]
Caused by: java.lang.Exception: Error starting bundle162: Activator start error in bundle NewKafkaArtifact [162].
    at org.apache.karaf.bundle.command.BundlesCommand.doExecute(BundlesCommand.java:66)[24:org.apache.karaf.bundle.core:4.0.3]
    ... 12 more
Caused by: org.osgi.framework.BundleException: Activator start error in bundle NewKafkaArtifact [162].
    at org.apache.felix.framework.Felix.activateBundle(Felix.java:2276)[org.apache.felix.framework-5.4.0.jar:]
    at org.apache.felix.framework.Felix.startBundle(Felix.java:2144)[org.apache.felix.framework-5.4.0.jar:]
    at org.apache.felix.framework.BundleImpl.start(BundleImpl.java:998)[org.apache.felix.framework-5.4.0.jar:]
    at org.apache.karaf.bundle.command.Start.executeOnBundle(Start.java:38)[24:org.apache.karaf.bundle.core:4.0.3]
    at org.apache.karaf.bundle.command.BundlesCommand.doExecute(BundlesCommand.java:64)[24:org.apache.karaf.bundle.core:4.0.3]
    ... 12 more
Caused by: org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found.
    at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:255)[141:kafka-examples:1.0.0.SNAPSHOT-jar-with-dependencies]
    at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:145)[141:kafka-examples:1.0.0.SNAPSHOT-jar-with-dependencies]
    at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:49)[141:kafka-examples:1.0.0.SNAPSHOT-jar-with-dependencies]
    at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:56)[141:kafka-examples:1.0.0.SNAPSHOT-jar-with-dependencies]
    at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerConfig.java:317)[141:kafka-examples:1.0.0.SNAPSHOT-jar-with-dependencies]
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:181)[141:kafka-examples:1.0.0.SNAPSHOT-jar-with-dependencies]
    at com.NewKafka.NewKafkaArtifact.KafkaProducerTest.main(KafkaProducerTest.java:25)[162:NewKafkaArtifact:0.0.1.SNAPSHOT]
    at com.NewKafka.NewKafkaArtifact.StartKafka.start(StartKafka.java:11)[162:NewKafkaArtifact:0.0.1.SNAPSHOT]
    at org.apache.felix.framework.util.SecureAction.startActivator(SecureAction.java:697)[org.apache.felix.framework-5.4.0.jar:]
    at org.apache.felix.framework.Felix.activateBundle(Felix.java:2226)[org.apache.felix.framework-5.4.0.jar:]
    ... 16 more

我试过设置 key.serializer 以及 value.serializer 如下所示：

props.put("key.serializer",StringSerializer.class.getName());
props.put("value.serializer",StringSerializer.class.getName());

也喜欢，但还是得到同样的错误。我做错什么了。

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
apache-kafkaapache-sparkspark-streaming-kafkaapache-karaf

6条答案

按热度按时间
yvgpqqbh

yvgpqqbh1#

试着用这些道具代替你的道具。

props.put("key.serializer", 
         "org.apache.kafka.common.serialization.StringSerializer");

  props.put("value.serializer", 
         "org.apache.kafka.common.serialization.StringSerializer");

这是Kafka的制作人example:-

import java.util.Properties; 
import org.apache.kafka.clients.producer.Producer;    
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class FxDateProducer {

   public static void main(String[] args) throws Exception{

      if(args.length == 0){
         System.out.println("Enter topic name”);
         return;
      }

      String topicName = args[0].toString(); 
      Properties props = new Properties();

      //Assign localhost id
      props.put("bootstrap.servers", “localhost:9092");

      //Set acknowledgements for producer requests.      
      props.put("acks", “all");

      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);

      //Specify buffer size in config
      props.put("batch.size", 16384);

      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);

      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);

      props.put("key.serializer", 
         "org.apache.kafka.common.serialization.StringSerializer");

      props.put("value.serializer", 
         "org.apache.kafka.common.serialization.StringSerializer");

      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);

      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(“Message sent successfully”);
               producer.close();
   }
}
赞(0）分享  回复(0）举报  2021-06-07
yptwkmov

yptwkmov2#

这是因为Kafka版本的问题。一定要使用正确的Kafka版本。我使用的版本是“Kafka2.12-1.0.1”
但请尝试在代码中使用以下属性。这解决了我的问题。

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

早些时候，我使用了下面的属性，这是导致问题的原因。

//props.put("key.serializer","org.apache.kafka.common.serialization.Stringserializer");
//props.put("value.serializer","org.apache.kafka.common.serialization.Stringserializer");
赞(0）分享  回复(0）举报  2021-06-07
z0qdvdin

z0qdvdin3#

我通过阅读Kafka客户端源代码找到原因。
Kafka客户端使用 Class.forName(trimmed, true, Utils.getContextOrKafkaClassLoader()) 要获取类对象并创建示例，关键点是classloader，它由方法的实现last param指定 Utils.getContextOrKafkaClassLoader() 是

public static ClassLoader getContextOrKafkaClassLoader() {
    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    if (cl == null)
        return getKafkaClassLoader();
    else
        return cl;
}

因此，在默认情况下 org.apache.kafka.common.serialization.StringSerializer 是由applicationclassloader加载的，如果你的目标类没有被applicationclassloader加载，这个问题就会发生！
要解决这个问题，只需在新的kafkaproducer示例之前将当前线程的contextclassloader设置为null，如下所示

Thread.currentThread().setContextClassLoader(null);
Producer<String, String> producer = new KafkaProducer(props);

希望我的回答能让你知道发生了什么。

赞(0）分享  回复(0）举报  2021-06-07
uoifb46i

uoifb46i4#

它与您使用的版本有关。还建议将其升级到版本0.8.2.2\u 1。建议您调整您正在使用的Kafka版本并试一试。代码方面，我交叉检查了kafka dev列表中的许多代码示例，似乎您的编写方式是正确的。
即 Thread.currentThread().setContextClassLoader(null);

赞(0）分享  回复(0）举报  2021-06-07
wbrvyc0a

wbrvyc0a5#

正如@ram ghadiyaram在他的回答中指出的那样，问题似乎与类装入器有关。为了让kafka客户机2.x使用此功能，我必须执行以下操作：

public Producer<String, String> createProducer() {
            ClassLoader original = Thread.currentThread().getContextClassLoader();
    Thread.currentThread().setContextClassLoader(null);
    Properties props = new Properties();

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
        BOOTSTRAP_SERVERS);
    ... etc ...

    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    Thread.currentThread().setContextClassLoader(original);
    return producer;

}

这允许系统继续使用原始类加载器加载其他类。这在wildfly/jboss中是需要的（我正在使用的特定应用程序是keydeap）。

赞(0）分享  回复(0）举报  2021-06-07
mcvgt66p

mcvgt66p6#

最近我找到了解决办法。将thead上下文加载器设置为null为我解决了这个问题。谢谢。

Thread.currentThread().setContextClassLoader(null);
Producer<String, String> producer = new KafkaProducer(props);
赞(0）分享  回复(0）举报  2021-06-07
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com