final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// checkpointing is required for exactly-once or at-least-once guarantees
env.enableCheckpointing(...);
final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
    .setHost("localhost")
    .setPort(5000)
    .build();
final DataStream<String> stream = env
    .addSource(new RMQSource<String>(
        connectionConfig,            // config for the RabbitMQ connection
        "queueName",                 // name of the RabbitMQ queue to consume
        true,                        // use correlation ids; can be false if only at-least-once is required
        new SimpleStringSchema()))   // deserialization schema to turn messages into Java objects
    .setParallelism(1);              // non-parallel source is only required for exactly-once
  Scala
  
val env = StreamExecutionEnvironment.getExecutionEnvironment
// checkpointing is required for exactly-once or at-least-once guarantees
env.enableCheckpointing(...)
val connectionConfig = new RMQConnectionConfig.Builder()
    .setHost("localhost")
    .setPort(5000)
    .build
val stream = env
    .addSource(new RMQSource[String](
        connectionConfig,            // config for the RabbitMQ connection
        "queueName",                 // name of the RabbitMQ queue to consume
        true,                        // use correlation ids; can be false if only at-least-once is required
        new SimpleStringSchema))     // deserialization schema to turn messages into Java objects
    .setParallelism(1)               // non-parallel source is only required for exactly-once
  Python
  
env = StreamExecutionEnvironment.get_execution_environment()
# checkpointing is required for exactly-once or at-least-once guarantees
env.enable_checkpointing(...)
connection_config = RMQConnectionConfig.Builder() \
    .set_host("localhost") \
    .set_port(5000) \
    .build()
stream = env \
    .add_source(RMQSource(
        connection_config,
        "queueName",
        True,
        SimpleStringSchema(),
    .set_parallelism(1)
  服务质量 (QoS) / 消费者预取(Consumer Prefetch)
RabbitMQ Source 通过 RMQConnectionConfig 类提供了一种简单的方式，来设置 source channel 上的 basicQos（见下方示例）。要注意的是这里的 prefetch count 是对单个 channel 设置的，并且由于每个并发的 source 都持有一个 connection/channel，因此这个值实际上会乘以 source 的并行度，来表示同一时间可以向这个 job 总共发送多少条未确认的消息。如果需要更复杂的配置，可以通过重写 RMQSource#setupChannel(Connection) 方法来实现手动配置。
  
final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
    .setPrefetchCount(30_000)
    .build();
  Scala
  
val connectionConfig = new RMQConnectionConfig.Builder()
    .setPrefetchCount(30000)
    .build
  Python
  
connection_config = RMQConnectionConfig.Builder() \
    .set_prefetch_count(30000) \
    .build()
RabbitMQ Source 默认情况下是不设置 prefetch count 的，这意味着 RabbitMQ 服务器将会无限制地向 source 发送消息。因此在生产环境中，最好要设置它。当消费海量数据的队列并且启用 checkpointing 时，消息只有在做完 checkpoint 后才会被确认，因此也许需要对 prefetch count 做一些调整来减少不必要的循环。
更多关于 QoS 以及 prefetch 相关的内容可以参考 这里.
更多关于在 AMQP 0-9-1 中可选的选项可以参考 这里.
  RabbitMQ Sink
该连接器提供了一个 RMQSink 类，用来向 RabbitMQ 队列发送数据。下面是设置 RabbitMQ sink 的代码示例：
  
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
  Scala
  
val stream: DataStream[String] = ...
val 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
  Python
  
stream = ...
connection_config = RMQConnectionConfig.Builder() \
    .set_host("localhost") \
    .set_port(5000) \
    .build()
stream.add_sink(RMQSink(
    connection_config,      # config for the RabbitMQ connection
    'queueName',            # name of the RabbitMQ queue to send messages to
    SimpleStringSchema()))  # serialization schema to turn Java objects to messages
更多关于 RabbitMQ 的信息请参考 这里.
 Back to top