java flink 发送rabbitMq 使用连接池

在使用 Flink 发送消息到 RabbitMQ 时，可以使用连接池来提高性能和资源利用率。下面是一个使用连接池发送消息到 RabbitMQ 的示例代码：

首先，你需要添加依赖项，以便使用连接池和 RabbitMQ 的 Java 客户端库：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-rabbitmq_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>${rabbitmq.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
        <version>${commons-pool2.version}</version>
    </dependency>
</dependencies>

然后，你可以使用 Flink 的 RabbitMQSink 来发送消息到 RabbitMQ。首先，你需要创建一个 RabbitMQ 连接工厂，然后将其传递给 RabbitMQSink。在这个过程中，你可以使用 Commons Pool2 库来创建和管理连接池。

import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.connector.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.connector.rabbitmq.common.RMQConnectionConfig.Builder;
import org.apache.flink.connector.rabbitmq.common.RMQConnectionConfig.RMQConnectionConfigBuilder;
import org.apache.flink.connector.rabbitmq.sink.RMQSink;
import com.rabbitmq.client.ConnectionFactory;

public class Main {
    public static void main(String[] args) throws Exception {
        // 创建 RabbitMQ 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");

        // 创建 RabbitMQ 连接配置
        RMQConnectionConfig connectionConfig = new RMQConnectionConfigBuilder()
                .setHost("localhost")
                .setPort(5672)
                .setUserName("guest")
                .setPassword("guest")
                .setVirtualHost("/")
                .build();

        // 创建连接池配置
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(10); // 设置连接池的最大连接数

        // 创建连接池
        GenericObjectPool<ConnectionFactory> pool = new GenericObjectPool<>(new ConnectionFactoryPoolFactory(factory), poolConfig);

        // 创建 RabbitMQSink
        RMQSink<String> sink = new RMQSink<>(connectionConfig, new SimpleStringSchema(), "exchangeName", "routingKey", pool);

        // 创建 Flink 程序
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> stream = env.fromElements("message1", "message2", "message3");
        stream.addSink(sink);

        // 执行 Flink 程序
        env.execute("RabbitMQ Sink Example");
    }

    // 连接工厂池工厂类
    private static class ConnectionFactoryPoolFactory extends BasePooledObjectFactory<ConnectionFactory> {
        private final ConnectionFactory factory;

        public ConnectionFactoryPoolFactory(ConnectionFactory factory) {
            this.factory = factory;
        }

        @Override
        public ConnectionFactory create() throws Exception {
            return factory.newConnection();
        }

        @Override
        public PooledObject<ConnectionFactory> wrap(ConnectionFactory obj) {
            return new DefaultPooledObject<>(obj);
        }

        @Override
        public void destroyObject(PooledObject<ConnectionFactory> p) throws Exception {
            p.getObject().close();
        }
    }
}

在上面的代码中，我们创建了一个 RabbitMQ 连接工厂，并使用它来创建 RabbitMQ 连接池。然后，我们使用连接池创建了一个 RMQSink，并将其传递给 Flink 程序的 DataStream 的 addSink 方法。

注意：在实际生产环境中，你可能需要根据需要调整连接池的配置，例如最大连接数、最大空闲连接数、连接超时时间等。

希望对你有所帮助