1

だから私は通知kafkaプロデューサーをkafkaコンシューマーに送信する単純なアプリケーションを実装したいと思います。これまでのところ、文字列メッセージをプロデューサーからコンシューマーに送信することに成功しました。利用した。

public class Notification implements Serializable{

    private String name;
    private String message;
    private long currentTimeStamp;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    public long getCurrentTimeStamp() {
        return currentTimeStamp;
    }

    public void setCurrentTimeStamp(long currentTimeStamp) {
        this.currentTimeStamp = currentTimeStamp;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        Notification that = (Notification) o;

        if (currentTimeStamp != that.currentTimeStamp) return false;
        if (message != null ? !message.equals(that.message) : that.message != null) return false;
        if (name != null ? !name.equals(that.name) : that.name != null) return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = name != null ? name.hashCode() : 0;
        result = 31 * result + (message != null ? message.hashCode() : 0);
        result = 31 * result + (int) (currentTimeStamp ^ (currentTimeStamp >>> 32));
        return result;
    }

    @Override
    public String toString() {
        return "Notification{" +
                "name='" + name + '\'' +
                ", message='" + message + '\'' +
                ", currentTimeStamp=" + currentTimeStamp +
                '}';
    }
}

そしてこちらがプロデューサー

public class KafkaProducer {
    static String topic = "kafka-tutorial";


    public static void main(String[] args) {
        System.out.println("Start Kafka producer");
        Properties properties = new Properties();
        properties.put("metadata.broker.list", "localhost:9092");
        properties.put("serializer.class", "dev.innova.kafka.tutorial.producer.CustomSerializer");
        ProducerConfig producerConfig = new ProducerConfig(properties);
        kafka.javaapi.producer.Producer<String, Notification> producer = new kafka.javaapi.producer.Producer<String, Notification>(producerConfig);
        KeyedMessage<String, Notification> message = new KeyedMessage<String, Notification>(topic, createNotification());
        System.out.println("send Message to broker");
        producer.send(message);
        producer.close();

    }

    private static Notification createNotification(){
        Notification notification = new Notification();
        notification.setMessage("Sample Message");
        notification.setName("Sajith");
        notification.setCurrentTimeStamp(System.currentTimeMillis());
        return notification;
    }
}

そして、これは消費者です

public class KafkaConcumer extends Thread {
    final static String clientId = "SimpleConsumerDemoClient";
    final static String TOPIC = "kafka-tutorial";
    ConsumerConnector consumerConnector;


    public KafkaConcumer() {
        Properties properties = new Properties();
        properties.put("zookeeper.connect","localhost:2181");
        properties.put("group.id","test-group");
        properties.put("serializer.class", "dev.innova.kafka.tutorial.producer.CustomSerializer");
        properties.put("zookeeper.session.timeout.ms", "400");
        properties.put("zookeeper.sync.time.ms", "200");
        properties.put("auto.commit.interval.ms", "1000");
        ConsumerConfig consumerConfig = new ConsumerConfig(properties);
        consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
    }

    @Override
    public void run() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(TOPIC, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream =  consumerMap.get(TOPIC).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        System.out.println("It :" + it.size());
        while(it.hasNext()){
            System.out.println(new String(it.next().message()));
        }
    }

    private static void printMessages(ByteBufferMessageSet messageSet) throws UnsupportedEncodingException {
        for(MessageAndOffset messageAndOffset: messageSet) {
            ByteBuffer payload = messageAndOffset.message().payload();
            byte[] bytes = new byte[payload.limit()];
            payload.get(bytes);
            System.out.println(new String(bytes, "UTF-8"));
        }
    }
}

そして最後に、 customserializer を使用してオブジェクトをシリアライズおよびデシリアライズしました。

public class CustomSerializer implements Encoder<Notification>, Decoder<Notification> {
    public CustomSerializer(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }
    @Override
    public byte[] toBytes(Notification o) {
        return new byte[0];
    }

    @Override
    public Notification fromBytes(byte[] bytes) {
        return null;
    }
}

誰かが問題を教えてもらえますか? これは正しい方法ですか？

4

3 に答える 3

3

2 つの問題があります。

まず、デシリアライザーにはロジックがありません。シリアル化するオブジェクトごとに空のバイト配列を返し、オブジェクトを逆シリアル化するように要求されるたびに null オブジェクトを返します。オブジェクトを実際にシリアライズおよびデシリアライズするコードをそこに配置する必要があります。

次に、JVM からネイティブ JVM シリアライゼーションおよびデシリアライゼーション ロジックを使用する場合は、転送される Bean に serialVersionUID を追加する必要があります。このようなもの：

private static final long serialVersionUID = 123L;

任意の値を使用できます。オブジェクトが JVM によってデシリアライズされると、オブジェクトの serialVersionId が、ロードされたクラス定義で指定された値と比較されます。2 つが異なる場合、JVM は、クラス定義がロードされていても、正しいバージョンのクラス定義がロードされておらず、シリアライゼーションが失敗すると想定します。クラス定義で serialVersionID の値を指定しない場合、JVM が 1 つを構成し、2 つの異なる JVM (プロデューサーとコンシューマーの 1 つ) がほぼ確実に異なる値を構成します。

編集

デフォルトの Java シリアライゼーションを活用したい場合は、シリアライザを次のようにする必要があります。

public class CustomSerializer implements Encoder<Notification>, Decoder<Notification> {
    public CustomSerializer(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }

@Override
public byte[] toBytes(Notification o) {
    try {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(baos);
        oos.writeObject(o);
        oos.close();
        byte[] b = baos.toByteArray();
        return b;
    } catch (IOException e) {
        return new byte[0];
    }
}

@Override
public Notification fromBytes(byte[] bytes) {
    try {
        return (Notification) new ObjectInputStream(new ByteArrayInputStream(b)).readObject();
    } catch (Exception e) {
        return null;
    }    
}
于 2015-11-30T12:07:57.587 に答える