java flink 发送rabbitMq消息 考虑性能使用连接池

在使用Java Flink发送RabbitMQ消息时，可以考虑使用连接池来提高性能。连接池可以在应用启动时创建一批RabbitMQ连接，并将其存储在连接池中。应用程序需要发送消息时，可以从连接池中获取一个连接，并将消息发送到RabbitMQ。

以下是使用连接池发送RabbitMQ消息的示例代码：

1. 创建连接池：

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

public class RabbitMQConnectionPool {
    private GenericObjectPool<Connection> connectionPool;

    public RabbitMQConnectionPool(String host, int port, String username, String password, int maxConnections) {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);

        GenericObjectPoolConfig<Connection> poolConfig = new GenericObjectPoolConfig<>();
        poolConfig.setMaxTotal(maxConnections);

        connectionPool = new GenericObjectPool<>(new RabbitMQConnectionFactory(connectionFactory), poolConfig);
    }

    public Connection getConnection() throws Exception {
        return connectionPool.borrowObject();
    }

    public void releaseConnection(Connection connection) {
        connectionPool.returnObject(connection);
    }
}

2. 创建连接工厂：

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;

public class RabbitMQConnectionFactory implements PooledObjectFactory<Connection> {
    private ConnectionFactory connectionFactory;

    public RabbitMQConnectionFactory(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    @Override
    public PooledObject<Connection> makeObject() throws Exception {
        return new DefaultPooledObject<>(connectionFactory.newConnection());
    }

    @Override
    public void destroyObject(PooledObject<Connection> pooledObject) throws Exception {
        Connection connection = pooledObject.getObject();
        connection.close();
    }

    @Override
    public boolean validateObject(PooledObject<Connection> pooledObject) {
        Connection connection = pooledObject.getObject();
        return connection.isOpen();
    }

    @Override
    public void activateObject(PooledObject<Connection> pooledObject) throws Exception {
        // Do nothing
    }

    @Override
    public void passivateObject(PooledObject<Connection> pooledObject) throws Exception {
        // Do nothing
    }
}

3. 使用连接池发送消息：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class RabbitMQProducer {
    private RabbitMQConnectionPool connectionPool;

    public RabbitMQProducer(RabbitMQConnectionPool connectionPool) {
        this.connectionPool = connectionPool;
    }

    public void sendMessage(String exchange, String routingKey, byte[] message) throws Exception {
        Connection connection = connectionPool.getConnection();
        Channel channel = connection.createChannel();

        channel.basicPublish(exchange, routingKey, null, message);

        channel.close();
        connectionPool.releaseConnection(connection);
    }
}

使用连接池发送RabbitMQ消息可以避免频繁地创建和关闭连接，从而提高性能和效率