rabbitmq - 无法在 Flink 1.3.2 中启动 RabbitMQ 源

Question

我想启动一个 RabbitMQ 源然后接收，但我无法执行第一步，即启动 Rabbit MQ 源。 RabbitMQ 服务器正在运行，我也可以看到仪表板。

我的代码如下

公共(public)类 rabbitmq_source {

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


    StreamExecutionEnvironment envrionment = StreamExecutionEnvironment.getExecutionEnvironment();


     RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
            .setHost("localhost")
            .setPort(50000).
                    setUserName("root")
             .setPassword("root").
                    setVirtualHost("/").build();


     DataStream<String> stream = envrionment
            .addSource(new RMQSource<String>(
                    connectionConfig,            // config for the RabbitMQ connection
                    "queue",                 // name of the RabbitMQ queue to consume
                    new SimpleStringSchema()));

    stream.print();

    envrionment.execute();



}

我不确定我应该设置什么用户名和密码，它们应该是 guest 还是 guest。但是，我收到以下错误

java.lang.RuntimeException: Cannot create RMQ connection with queue at localhost
    at org.apache.flink.streaming.connectors.rabbitmq.RMQSource.open(RMQSource.java:172)
    at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:111)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:376)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:253)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.net.ConnectException: Connection refused (Connection refused)
    at java.net.PlainSocketImpl.socketConnect(Native Method)
    at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
    at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
    at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
    at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
    at java.net.Socket.connect(Socket.java:589)
    at com.rabbitmq.client.impl.FrameHandlerFactory.create(FrameHandlerFactory.java:32)
    at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:588)
    at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:612)

Answer 1

将 LocalStreamEnvironment.createLocalEnvironment() 与 guest 用户名和密码一起使用。