在 Flink SourceFunction 中获取 ClassNotFound 异常

Question

我正在使用协议缓冲区将数据流发送到 Apache Flink。我有两节课。一种是生产者，一种是消费者。Producer 是一个 java 线程类，它从 socket 中读取数据，Protobuf 将其反序列化，然后将其存储在我的 BlockingQueue 中。 Consumer 是一个在 Flink 中实现 SourceFunction 的类。我使用以下方法测试了这个程序：

DataStream<Event.MyEvent> stream = env.fromCollection(queue);

而不是自定义源，它工作正常。但是当我尝试使用 SourceFunction 类时，它会引发此异常：

Caused by: java.lang.RuntimeException: Unable to find proto buffer class
at com.google.protobuf.GeneratedMessageLite$SerializedForm.readResolve(GeneratedMessageLite.java:775)
...
Caused by: java.lang.ClassNotFoundException: event.Event$MyEvent
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
...

在另一次尝试中，我将两个分类为一个（实现 SourceFunction 的类）。我从套接字获取数据并使用 protobuf 将其反序列化并将其存储在 BlockingQueue 中，然后我立即从 BlockingQueue 中读取。我的代码也适用于这种方法。

但我想使用两个单独的类（多线程），但它会抛出该异常。我试图在过去 2 天内解决它，也做了很多搜索，但没有运气。任何帮助都会受到重视。

制作人：

public class Producer implements Runnable {

    Boolean running = true;
    Socket socket = null, bufferSocket = null;
    PrintStream ps = null;
    BlockingQueue<Event.MyEvent> queue;
    final int port;

    public Producer(BlockingQueue<Event.MyEvent> queue, int port){
        this.port = port;
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            socket = new Socket("127.0.0.1", port);
            bufferSocket = new Socket(InetAddress.getLocalHost(), 6060);
            ps = new PrintStream(bufferSocket.getOutputStream());
            while (running) {
                queue.put(Event.MyEvent.parseDelimitedFrom(socket.getInputStream()));
                ps.println("Items in Queue: " + queue.size());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

消费者：

public class Consumer implements SourceFunction<Event.MyEvent> {

    Boolean running = true;
    BlockingQueue<Event.MyEvent> queue;
    Event.MyEvent event;
    public Consumer(BlockingQueue<Event.MyEvent> queue){
        this.queue = queue;
    }

    @Override
    public void run(SourceContext<Event.MyEvent> sourceContext) {
        try {
            while (running) {
                event = queue.take();
                sourceContext.collect(event);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}

Event.MyEvent 是我的 protobuf 类。我使用的是 2.6.1 版，我用 v2.6.1 编译了类。我仔细检查了版本以确保这不是问题。Producer 类工作正常。我用 Flink v1.1.3 和 v1.1.4 对此进行了测试。我在本地模式下运行它。

---

编辑：答案包含在问题中，单独发布并在此处删除。

更新 12/28/2016

……但我还是很好奇。是什么导致了这个错误？这是 Flink 中的错误还是我做错了什么？

...

Answer 1

提问者已经找到了一种方法来实现这一点。我已经从问题中提取了相关部分。请注意，发生这种情况的原因仍然无法解释。

我没有使用引号语法，因为它有很多文本，但提问者分享了以下内容：

所以最后我让它工作了。我在 SourceFunction （消费者）内部创建了 BlockingQueue 对象，并从 SourceFunction 类（消费者）内部调用 Producer 类，而不是在程序的 main 方法中创建 BlockingQueue 并调用 Producer 类。现在可以了！

这是我在 Flink 中的完整工作代码：

public class Main {

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

    final int port, buffer;
    //final String ip;
    try {
        final ParameterTool params = ParameterTool.fromArgs(args);
        port = params.getInt("p");
        buffer = params.getInt("b");
    } catch (Exception e) {
        System.err.println("No port number and/or buffer size specified.");
        return;
    }

    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


    DataStream<Event.MyEvent> stream = env.addSource(new Consumer(port, buffer));
    //DataStream<Event.MyEvent> stream = env.fromCollection(queue);


    Pattern<Event.MyEvent, ?> crashedPattern = Pattern.<Event.MyEvent>begin("start")
            .where(new FilterFunction<Event.MyEvent>() {
                @Override
                public boolean filter(Event.MyEvent myEvent) throws Exception {
                    return (myEvent.getItems().getValue() >= 120);
                }
            })
            .<Event.MyEvent>followedBy("next").where(new FilterFunction<Event.MyEvent>() {
                @Override
                public boolean filter(Event.MyEvent myEvent) throws Exception {
                    return (myEvent.getItems().getValue() <= 10);
                }
            })
            .within(Time.seconds(3));

    PatternStream<Event.MyEvent> crashed = CEP.pattern(stream.keyBy(new KeySelector<Event.MyEvent, String>() {
        @Override
        public String getKey(Event.MyEvent myEvent) throws Exception {
            return myEvent.getEventType();
        }
    }), crashedPattern);

    DataStream<String> alarm = crashed.select(new PatternSelectFunction<Event.MyEvent, String>() {
        @Override
        public String select(Map<String, Event.MyEvent> pattern) throws Exception {
            Event.MyEvent start = pattern.get("start");
            Event.MyEvent next = pattern.get("next");
            return start.getEventType() + " | Speed from " + start.getItems().getValue() + " to " + next.getItems().getValue() + " in 3 seconds\n";
        }
    });

    DataStream<String> rate = alarm.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
            .apply(new AllWindowFunction<String, String, TimeWindow>() {
                @Override
                public void apply(TimeWindow timeWindow, Iterable<String> iterable, Collector<String> collector) throws Exception {
                    int sum = 0;
                    for (String s: iterable) {
                        sum ++;
                    }
                    collector.collect ("CEP Output Rate: " + sum + "\n");
                }
            });

    rate.writeToSocket(InetAddress.getLocalHost().getHostName(), 7070, new SimpleStringSchema());

    env.execute("Flink Taxi Crash Streaming");
}

private static class Producer implements Runnable {

    Boolean running = true;
    Socket socket = null, bufferSocket = null;
    PrintStream ps = null;
    BlockingQueue<Event.MyEvent> queue;
    final int port;

    Producer(BlockingQueue<Event.MyEvent> queue, int port){
        this.port = port;
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            socket = new Socket("127.0.0.1", port);
            bufferSocket = new Socket(InetAddress.getLocalHost(), 6060);
            ps = new PrintStream(bufferSocket.getOutputStream());
            while (running) {
                queue.put(Event.MyEvent.parseDelimitedFrom(socket.getInputStream()));
                ps.println("Items in Queue: " + queue.size());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

private static class Consumer implements SourceFunction<Event.MyEvent> {

    Boolean running = true;
    final int port;
    BlockingQueue<Event.MyEvent> queue;

    Consumer(int port, int buffer){
        queue = new ArrayBlockingQueue<>(buffer);
        this.port = port;
    }

    @Override
    public void run(SourceContext<Event.MyEvent> sourceContext) {
        try {
            new Thread(new Producer(queue, port)).start();
            while (running) {
                sourceContext.collect(queue.take());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}