Flink RMQSink寫入RabbitMQ出現Connection數一直持續增加的問題

因業務需求，Flink程式要往RabbitMQ中寫資料。一開始使用自己開發環境下的RabbitM，那是非常的順溜。到測試環境，就出問題了。在RabbitMQ的web ui頁面，看到連接配接數猛的增加，而且一直不停。導緻測試環境的RabbitMQ挂了。當時蒙圈啊，出啥問題了，這麼嚴重。

經過一番排查對比。發現在

RMQSink

類中，在

open()

方法中，調用了一個方法

setupQueue()

這個方法的實作：

/**
* Sets up the queue. The default implementation just declares the queue. The user may override
* this method to have a custom setup for the queue (i.e. binding the queue to an exchange or
* defining custom queue parameters)
*/
protected void setupQueue() throws IOException {
    if (queueName != null) {
    	channel.queueDeclare(queueName, false, false, false, null);
    }
}
           

如果隊列名不為空，那麼執行語句。這個

queueDeclare()

的第二個參數是

durable

。隊列是否持久化。false為不支援。一般在我們寫入RabbitMQ之前，RabbitMQ的對列已經建立好，有些會指定

durable

為true。這個時候就有問題了，因為RabbitMQ隊列

durable

屬性不同，就會一直去連接配接，直到RabbitMQ挂掉或程式關掉。

解決這個bug，我們隻需要重寫

RMQSink

的

setupQueue()

方法即可。下面的代碼實作：

public class RichRMQSink extends RMQSink<String> {
    public RichRMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<String> schema) {
        super(rmqConnectionConfig, queueName, schema);
    }

    /**
     * 此方法必須重寫，如果RabbitMQ的queue的durable屬性設定為true，則會導緻RabbitMQ會一直connection，導緻RabbitMQ耗盡資源挂掉
     */
    @Override
    protected void setupQueue() {

    }
}
           

下面是RabbitMQ sink的代碼：

...
operator = env.addSource()
...

// 寫入RabbitMQ
RMQConnectionConfig rmqConnectionConfig = new RMQConnectionConfig.Builder()
        .setHost(Constant.SINK_RABBITMQ_HOST)
        .setVirtualHost(Constant.SINK_RABBITMQ_VIRTUAL_HOST)
        .setPort(5672)
        .setUserName(Constant.SINK_RABBITMQ_USERNAME)
        .setPassword(Constant.SINK_RABBITMQ_PASSWARD)
        .build();

String queueName = Constant.SINK_RABBITMQ_QUEUENAME;
// queueName 需要先在RabbitMQ建立好
operator.addSink(new RichRMQSink(rmqConnectionConfig, queueName, new SimpleStringSchema()));

env.executor()