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()