Flink 提供了一系列常见的内建连接器，用于支持各类系统的 DataSources 数据导入、Sinks 数据导出。
以下是 Apache RabbitMQ Connectors 的使用示例：
/* RabbitMQ DataSources 使用 */
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置 checkpoint 担保
env.enableCheckpointing(...);
final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
.setHost("localhost")
.setPort(5000)
...
.build();
final DataStream<String> stream = env
.addSource(new RMQSource<String>(
connectionConfig, // 配置 RabbitMQ 连接
"queueName", // 连接的 RabbitMQ 队列名称
true, // use correlation ids; can be false if only at-least-once is required
new SimpleStringSchema())) // 序列化，message 消息转为 Java Object
.setParallelism(1); // non-parallel source is only required for exactly-once
/* RabbitMQ Sinks 使用 */
final DataStream<String> stream = ...
final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
.setHost("localhost")
.setPort(5000)
...
.build();
stream.addSink(new RMQSink<String>(
connectionConfig, // config for the RabbitMQ connection
"queueName", // name of the RabbitMQ queue to send messages to
new SimpleStringSchema())); // serialization schema to turn Java objects to messages