2

因此，API 0.8.2 中生产者的 Kakfa 配置属性发生了变化；在过去工作并让我的 java Producer 编译之后，我得到了一个异常。生产者针对我的 Kafka_2.9.1-0.8.2.1 集群的节点，我得到了关于DefaultSerializer未实例化的异常：

Exception in thread "main" org.apache.kafka.common.KafkaException: Could not instantiate class kafka.serializer.DefaultEncoder Does it have a public no-argument constructor?
        at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:235)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:136)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:216)
........

鉴于这是在 Kakfa 中实现的，我想知道将 Kafka 作为依赖项进行编译是否不够，因为我可能需要在运行时打包一个或多个 Kafka jar。我还没有找到关于此的文档（最新的或其他的）。是否有我缺少的生产者运行时 jar？

作为参考，我在build.gradle这里包括了我的（有点乱）。编译中的排除项是在已经收到此错误之后的新增内容，因此无论是否依赖项块中的这些行都会发生错误。我确实尝试只依赖kafka-client0.8.2 的模块，但我认为这不适用于制作人。这是文件：

buildscript {
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath 'com.google.protobuf:protobuf-gradle-plugin:0.7.0'
    }
}

group 'lamblin'
version '0.1-SNAPSHOT'

apply plugin: 'application'
apply plugin: "com.google.protobuf"

sourceCompatibility = 1.7
targetCompatibility = 1.7

// Eliminates bootstrap class warning from javac
//tasks.withType(Compile) {
//    options.bootClasspath = "$JDK6_HOME/jre/lib/rt.jar"
//}

repositories {
    mavenCentral()
}

dependencies {
    testCompile group: 'junit', name: 'junit', version: '4.11'
    compile group: 'com.google.guava', name: 'guava', version: '18.0'
    compile group: 'com.google.protobuf', name: 'protobuf-java-util', version: '3.0.0-beta-1'
    compile group: 'com.google.transit', name: 'gtfs-realtime-bindings', version: '0.0.4'
    compile group: 'com.offbytwo', name: 'docopt', version: '0.6.0.20150202'
    //compile group: 'org.apache.kafka', name: 'kafka_2.9.1', version: '0.8.2.1' {
    compile ('org.apache.kafka:kafka_2.9.1:0.8.2.1') {
        exclude group: 'com.sun.jmx', module: 'jmxri'
        exclude group: 'javax.jmx', module: 'jms'
        exclude group: 'com.sun.jdmk', module: 'jmxtools'
    }
}

protobuf {
    generateProtoTasks {
        all().each { task ->
            task.builtins {
                python { }
            }
        }
    }
    protoc {
        //  artifact = 'com.google.protobuf:protoc:3.0.0-alpha-3'
        artifact = 'com.google.protobuf:protoc:2.6.1'
    }
}
// First Application Script
mainClassName = "com.insight.lamblin.GtfsToJson"
applicationName = "gtfsToJson"

// Subsequent Scripts
task createAllStartScripts() << {
    // This task is added to by a loop over the scripts array creating-sub tasks
}
def scripts = [ 'gtfsToJson': 'com.insight.lamblin.GtfsToJson',
                'rawGtfsKafkaProducer': 'com.insight.lamblin.RawGtfsKafkaProducer'
]
scripts.each() { scriptName, className ->
    def t = tasks.create(name: scriptName+'StartScript', type: CreateStartScripts) {
        mainClassName = className
        applicationName = scriptName
        outputDir = new File(project.buildDir, 'scripts')
        classpath = jar.outputs.files + project.configurations.runtime
    }
    applicationDistribution.into("bin") {
        from(t)
        fileMode = 0755
    }
    createAllStartScripts.dependsOn(t)
}
4

1 回答 1

2

场景：不起眼的教堂地下室里，一圈折叠的金属椅子被各种中年男人和少数女人占据，其中大多数人戴着眼镜，显得不感兴趣。一盒咖啡和一些半甜甜圈依次排列在塑料盘子上，放在靠近入口的墙边的一张有缺口的、裸露的、折叠的桌子上。

丹尼尔：嗨，我叫丹尼尔，我是一个……（呜咽）……文件浏览器。
小组（慢慢地）：欢迎丹尼尔。

写那个剧本是因为我的卡夫卡问题似乎只吸引了蟋蟀，所以我在这里让它有点有趣……感觉很孤独。

在我的辩护中，有 10 个看似权威的 kafka.apache.org 文档关于生产者的属性设置。这kafka.serializer.DefaultSerializer在几乎所有设置属性的示例中都非常突出和常见，Java 生产者示例完全缺乏有关属性或运行示例代码的详细信息。

此外，尽管名称为“default”，但该属性没有默认值，因此需要设置它。这似乎是一个愚蠢的细节，但对于 Kafka 开发团队的某些人来说，它一定是有意义的。

在运行用 Java 编写的 Kafka 生产者时，生产者应该从少数可用的 Java 特定编码器中设置编码器。前面提到的似乎是 Scala 特有的。对于您感兴趣的 Java org.apache.kafka.common.serialization，您等效于默认序列化程序的是：ByteArraySerializer. 如果你设置key.serializer，value.serializer它应该工作。设置它们的更好方法是使用ProducerConfiglike中的静态字符串ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG。

设置有点像：

import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.ProducerConfig; 
import org.apache.kafka.clients.producer.ProducerRecord; 
... 
import java.util.Properties;
...
Properties props = new Properties(); 
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
          "172.31.22.7:9092,172.31.22.6:9092,172.31.22.5:9092,172.31.22.4:9092"); 
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
          "org.apache.kafka.common.serialization.ByteArraySerializer"); 
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
          "org.apache.kafka.common.serialization.ByteArraySerializer"); 
KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props); 
...
于 2015-10-05T14:33:22.220 回答