docker run -d  -p 15672:15672  -p  5672:5672  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin --name rabbitmq rabbitmq:3-management

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-rabbitmq_${scala.binary.version}</artifactId><version>${flink.version}</version>
</dependency>

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class RabbitMQProducerUtil {public final static String QUEUE_NAME = "zhisheng";public static void main(String[] args) throws Exception {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();//设置RabbitMQ相关信息factory.setHost("localhost");factory.setUsername("admin");factory.setPassword("admin");factory.setPort(5672);//创建一个新的连接Connection connection = factory.newConnection();//创建一个通道Channel channel = connection.createChannel();// 声明一个队列
//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);//发送消息到队列中String message = "Hello zhisheng";//我们这里演示发送一千条数据for (int i = 0; i < 1000; i++) {channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes("UTF-8"));System.out.println("Producer Send +'" + message + i);}//关闭通道和连接channel.close();connection.close();}
}

import com.zhisheng.common.utils.ExecutionEnvUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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;/*** 从 rabbitmq 读取数据*/
public class Main {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();ParameterTool parameterTool = ExecutionEnvUtil.PARAMETER_TOOL;//这些配置建议可以放在配置文件中，然后通过 parameterTool 来获取对应的参数值final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder().setHost("localhost").setVirtualHost("/").setPort(5672).setUserName("admin").setPassword("admin").build();DataStreamSource<String> zhisheng = env.addSource(new RMQSource<>(connectionConfig,"zhisheng",true,new SimpleStringSchema())).setParallelism(1);zhisheng.print();//如果想保证 exactly-once 或 at-least-once 需要把 checkpoint 开启
//        env.enableCheckpointing(10000);env.execute("flink learning connectors rabbitmq");}
}

nested exception is com.rabbitmq.client.AuthenticationFailureException: ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile.

// 声明一个队列
//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'zhisheng' in vhost '/': received 'true' but current is 'false', class-id=50, method-id=10)

public class Main1 {public static void main(String[] args) throws Exception {final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);StreamExecutionEnvironment env = ExecutionEnvUtil.prepare(parameterTool);DataStreamSource<Metrics> data = KafkaConfigUtil.buildSource(env);final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder().setHost("localhost").setVirtualHost("/").setPort(5672).setUserName("admin").setPassword("admin").build();//注意，换一个新的 queue，否则也会报错data.addSink(new RMQSink<>(connectionConfig, "zhisheng001", new MetricSchema()));env.execute("flink learning connectors rabbitmq");}
}