相关疑难解决方法(0)

我想不可能从Java调用Scala特征中实现的方法,或者有办法吗？

假设我在Scala中:

trait Trait {
  def bar = {}
}

如果我使用它,在Java中

class Foo implements Trait {
}

Java抱怨说 Trait is not abstract and does not override abstract method bar() in Trait

我是Kafka,Serialization和JSON的新手

我想要的是生产者通过kafka和消费者发送JSON文件以使用原始文件形式的JSON文件.

我能够得到它所以JSON转换为字符串并通过String Serializer发送然后消费者将解析String并重新创建一个JSON对象但我担心这不是有效的或正确的方法(可能会失去字段类型对于JSON)

所以我研究了制作JSON序列化程序并在我的制作人的配置中设置它.

我在这里使用了JsonEncoder:Kafka:编写自定义序列化器

但是当我现在尝试运行我的生产者时,似乎在编码器的toBytes函数中,try块永远不会返回任何像我想要的那样

try {
            bytes = objectMapper.writeValueAsString(object).getBytes();

        } catch (JsonProcessingException e) {
            logger.error(String.format("Json processing failed for object: %s", object.getClass().getName()), e);
        }

似乎objectMapper.writeValueAsString(object).getBytes(); 接受我的JSON obj({"name":"Kate","age":25})并将其转换为空,

这是我的制作人的跑步功能

List<KeyedMessage<String,JSONObject>> msgList=new ArrayList<KeyedMessage<String,JSONObject>>();   

    JSONObject record = new JSONObject();

    record.put("name", "Kate");
    record.put("age", 25);

    msgList.add(new KeyedMessage<String, JSONObject>(topic, record));

    producer.send(msgList);

我错过了什么？我的原始方法(转换为字符串并发送然后重建JSON obj)是否可以？或者只是没有正确的方法去？

谢谢!

只有很少的序列化器可用,如,

org.apache.kafka.common.serialization.StringSerializer
org.apache.kafka.common.serialization.StringSerializer 

我们如何创建自己的自定义序列化程序？

我开始和卡夫卡一起玩.我已经设置了一个zookeeper配置,我设法发送和使用String消息.现在我试图传递一个Object(在java中),但由于某种原因,在消费者中解析Message时我有头问题.我尝试了几个序列化选项(使用Decoder/Encoder),并且所有返回相同的头问题.

这是我的代码生产者:

        Properties props = new Properties();
        props.put("zk.connect", "localhost:2181");
        props.put("serializer.class", "com.inneractive.reporter.kafka.EventsDataSerializer");
        ProducerConfig config = new ProducerConfig(props);
        Producer<Long, EventDetails> producer = new Producer<Long, EventDetails>(config);
        ProducerData<Long, EventDetails> data = new ProducerData<Long, EventDetails>("test3", 1, Arrays.asList(new EventDetails());
        try {
           producer.send(data);
        } finally {
           producer.close();
        }

和消费者:

        Properties props = new Properties();
        props.put("zk.connect", "localhost:2181");
        props.put("zk.connectiontimeout.ms", "1000000");
        props.put("groupid", "test_group");

        // Create the connection to the cluster
        ConsumerConfig consumerConfig = new ConsumerConfig(props);
        ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);

        // create 4 partitions of the stream for topic “test”, to …