如何将ApacheFlink连接到rabbitmq?

ccrfmcuu  于 2021-06-24  发布在  Flink
关注(0)|答案(1)|浏览(436)

尝试将apache flink连接到rabbitmq时出错。我用apachemaven做了这个项目,但是错误仍然存在。在这个项目中,我使用from mvn clean包进行编译,但我有一个错误。后来我使用from idea编译器,但我有一个错误。
我的代码是:

package org.apache.flink;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
class App
{
public static void main( String[] args )
{
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.enableCheckpointing(10);
    final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
            .setHost("local")
            .setPort(5000)
            .build();
    final DataStream<String> stream = env
            .addSource(new RMQSource<String>(
                    connectionConfig,            
                    "queueName",                 
                    true,                        
                    new SimpleStringSchema()))   
            .setParallelism(1);             
}
}

我的pom是:

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rabbitmq_2.12</artifactId>
<version>1.8.1</version>

<name>flink-connector-rabbitmq_2.12</name>
<!-- FIXME change it to the project's website -->
<url>http://www.example.com</url>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.7</maven.compiler.source>
<maven.compiler.target>1.7</maven.compiler.target>
</properties>

<dependencies>
<dependency>
  <groupId>junit</groupId>
  <artifactId>junit</artifactId>
  <version>4.11</version>
  <scope>test</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-rabbitmq_2.11</artifactId>
  <version>1.8.1</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${project.version}</version>
  <scope>provided</scope>
</dependency>

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.7.3</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  <type>test-jar</type>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime_2.11</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${project.version}</version>
  <scope>provided</scope>
</dependency>

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.7.3</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  <type>test-jar</type>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime_2.11</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-core</artifactId>
  <version>1.8.1</version>
</dependency>

</dependencies>

<build>
<pluginManagement><!-- lock down plugins versions to avoid using Maven    defaults (may be moved to parent pom) -->
  <plugins>
    <!-- clean lifecycle, see https://maven.apache.org/ref/current  /maven-core/lifecycles.html#clean_Lifecycle -->
    <plugin>
      <artifactId>maven-clean-plugin</artifactId>
      <version>3.1.0</version>
    </plugin>
    <!-- default lifecycle, jar packaging: see   https://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging -->
    <plugin>
      <artifactId>maven-resources-plugin</artifactId>
      <version>3.0.2</version>
    </plugin>
    <plugin>
      <artifactId>maven-compiler-plugin</artifactId>
      <version>3.8.0</version>
    </plugin>
    <plugin>
      <artifactId>maven-surefire-plugin</artifactId>
      <version>2.22.1</version>
    </plugin>
    <plugin>
      <artifactId>maven-jar-plugin</artifactId>
      <version>3.0.2</version>
    </plugin>
    <plugin>
      <artifactId>maven-install-plugin</artifactId>
      <version>2.5.2</version>
    </plugin>
    <plugin>
      <artifactId>maven-deploy-plugin</artifactId>
      <version>2.8.2</version>
    </plugin>
    <!-- site lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#site_Lifecycle -->
    <plugin>
      <artifactId>maven-site-plugin</artifactId>
      <version>3.7.1</version>
    </plugin>
    <plugin>
      <artifactId>maven-project-info-reports-plugin</artifactId>
      <version>3.0.0</version>
    </plugin>
    </plugins>
   </plugin Management>
   </build>
   </project>

当我运行程序时,我遇到以下错误
我的错误是:

Error: A JNI error has occurred, please check your installation and  try again
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache /flink/streaming/api/functions/source/SourceFunction
at java.lang.Class.getDeclaredMethods0(Native Method)
at java.lang.Class.privateGetDeclaredMethods(Class.java:2701)
at java.lang.Class.privateGetMethodRecursive(Class.java:3048)
at java.lang.Class.getMethod0(Class.java:3018)
at java.lang.Class.getMethod(Class.java:1784)
at   sun.launcher.LauncherHelper.validateMainClass(LauncherHelper.java:544)
at  sun.launcher.LauncherHelper.checkAndLoadMain(LauncherHelper.java:526)
Caused by: java.lang.ClassNotFoundException:   org.apache.flink.streaming.api.functions.source.SourceFunction
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 7 more

请帮帮我

arknldoa

arknldoa1#

这里的问题是,你有你的Flink依赖 provided scope,因为通常情况下,如果您在flink集群上运行应用程序,flink二进制文件已经存在,因此您不需要将它们包含在fat jar中,因此它们在scope中 provided . 但是我假设您正在本地运行作业,因此flink依赖项不可用于编译,在这种情况下,您需要更改范围,以便这些依赖项在编译时可用。如果你改变主意 scope 或者把它从 pom.xml (默认范围为 compile )它应该很好用。

相关问题