import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
public static Connection createConnection(final ProcessContext context) throws IOException, TimeoutException {

        Config config = new Config();
        final String rabbitHost = context.getProperty(RABBITMQ_HOST).getValue();
        final int rabbitPort = context.getProperty(RABBITMQ_PORT).asInteger();
        final String rabbitVirtualHost = context.getProperty(RABBITMQ_VIRTUALHOST).getValue();
        final String rabbitUsername = context.getProperty(RABBITMQ_USERNAME).getValue();
        final String rabbitPassword = context.getProperty(RABBITMQ_PASSWORD).getValue();


        config = config.withRecoveryPolicy(RecoveryPolicies.recoverAlways())
                .withRetryPolicy(new RetryPolicy()
                        .withMaxAttempts(200)
                        .withInterval(Duration.seconds(1))
                        .withMaxDuration(Duration.minutes(5)));

        ConnectionOptions options = new ConnectionOptions()
                .withHost(rabbitHost)
                .withPort(rabbitPort)
                .withVirtualHost(rabbitVirtualHost)
                .withUsername(rabbitUsername)
                .withPassword(rabbitPassword);

        return Connections.create(options, config);
    }
 

import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
public static Connection createAmqpConnection(final ExecutorService executor, Config config) throws IOException
{
    final ConnectionOptions options = new ConnectionOptions()
            .withConsumerExecutor(executor)
            .withPort(config.getInt("amqp.port"))
            .withHost(config.getString("amqp.host"))
            .withUsername(config.getString("amqp.user"))
            .withPassword(config.getString("amqp.pass"))
            .withVirtualHost(config.getString("amqp.vhost"));

    final net.jodah.lyra.config.Config c = new net.jodah.lyra.config.Config()
        .withRecoveryPolicy(RecoveryPolicies.recoverAlways())
        .withRetryPolicy(new RetryPolicy().withBackoff(Duration.seconds(1), Duration.seconds(30)));

    return Connections.create(options, c);
}
 

import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
public static Connection createRabbitmqConnection(String host, int port, String user, String password) throws IOException, TimeoutException {
    Config config = new Config()
            .withRecoveryPolicy(new RecoveryPolicy()
                    .withBackoff(Duration.seconds(1), Duration.seconds(30))
                    .withMaxAttempts(20));
    ConnectionOptions options = new ConnectionOptions()
            .withHost(host)
            .withPort(port)
            .withUsername(user)
            .withPassword(password);

    return Connections.create(options, config);
}
 

import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
@PostConstruct
public void start() throws IOException, TimeoutException {
    // millis
    connectionFactory.setConnectionTimeout(1000);
    // seconds
    connectionFactory.setRequestedHeartbeat(4);
    // lyra reconnect logic
    Config config = new Config()
            .withRecoveryPolicy(new RecoveryPolicy()
                    .withMaxAttempts(-1)
                    .withInterval(Duration.seconds(1)))
            .withChannelListeners(this);

    ConnectionOptions connectionOptions = new ConnectionOptions(connectionFactory)
            .withHosts(StringUtils.commaDelimitedListToStringArray(rabbitmqHosts))
            .withPort(rabbitmqPort)
            .withUsername(username)
            .withPassword(password);
    // create single connection
    //clientConnection = connectionFactory.newConnection(Address.parseAddresses(rabbitmqHosts));
    clientConnection = Connections.create(connectionOptions,config);
    // create a seperate producer and a seperate consumer channel
    consumerChannel = clientConnection.createChannel();
    consumerChannel.basicQos(prefetchCount);
    producerChannel = clientConnection.createChannel();
    // ensure the exchange is there
    consumerChannel.exchangeDeclare(exchangeName,"direct",true);
    if(ackType == BUFFERED) {
        messageAcker = new BufferingMessageAcker(consumerChannel);
    } else if(ackType == WRITE_BEHIND) {
        messageAcker = new WriteBehindMessageAcker(consumerChannel);
    } else if(ackType == ASYNC) {
        messageAcker = new AsyncMessageAcker(consumerChannel);
    } else {
        messageAcker = new DirectMessageAcker(consumerChannel);
    }
    messageAcker.start();
}
 

import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
public ConnectionHandler(ConnectionOptions options, Config config, ClassLoader classLoader) throws IOException {
  this.options = options;
  this.config = config;
  this.classLoader = Assert.notNull(classLoader, "classLoader");
  this.connectionName =
      options.getName() == null ? String.format("cxn-%s", CONNECTION_COUNTER.incrementAndGet())
          : options.getName();
  consumerThreadPool =
      options.getConsumerExecutor() == null ? Executors.newCachedThreadPool(new NamedThreadFactory(
          String.format("rabbitmq-%s-consumer", connectionName), config.isUsingDaemonThreads())) : options.getConsumerExecutor();
}
 

import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
protected void mockConnection() throws IOException, TimeoutException {
  if (connectionFactory == null) {
    mockConnectionOnly();
    connectionFactory = mock(ConnectionFactory.class);
    when(connectionFactory.getVirtualHost()).thenReturn("/");
    when(connectionFactory.newConnection(any(ExecutorService.class), any(Address[].class), anyString()))
        .thenReturn(connection);
  }

  if (options == null)
    options = new ConnectionOptions().withHost("test-host");
  options.withConnectionFactory(connectionFactory);
  if (config == null)
    config =
        new Config().withRetryPolicy(
            RetryPolicies.retryAlways().withInterval(Duration.millis(10))).withRecoveryPolicy(
            RecoveryPolicies.recoverAlways());

  if (connectionHandler == null) {
    connectionHandler = new ConnectionHandler(options, config, Connection.class.getClassLoader());
    connectionProxy =
        (ConfigurableConnection) Proxy.newProxyInstance(Connection.class.getClassLoader(),
            new Class<?>[] {ConfigurableConnection.class}, connectionHandler);
    connectionHandler.createConnection(connectionProxy);
    channels = new HashMap<Integer, MockChannel>();
  }
}
 

import net.jodah.lyra.ConnectionOptions; //导入依赖的package包/类
@PostConstruct
public void start() throws IOException, TimeoutException {
    // millis
    connectionFactory.setConnectionTimeout(1000);
    // seconds
    connectionFactory.setRequestedHeartbeat(4);
    // lyra reconnect logic
    Config config = new Config()
            .withRecoveryPolicy(new RecoveryPolicy()
                    .withMaxAttempts(-1)
                    .withInterval(Duration.seconds(1)))
            .withChannelListeners(this);

    ConnectionOptions connectionOptions = new ConnectionOptions(connectionFactory)
            .withHosts(StringUtils.commaDelimitedListToStringArray(rabbitmqHosts))
            .withPort(rabbitmqPort)
            .withUsername(username)
            .withPassword(password);
    // create single connection
    //clientConnection = connectionFactory.newConnection(Address.parseAddresses(rabbitmqHosts));
    clientConnection = Connections.create(connectionOptions,config);
    // create a seperate consumer channel
    consumerChannel = clientConnection.createChannel();
    consumerChannel.basicQos(prefetchCount);
    // prepare the consumer channels
    for (int i = 0; i < queueExecutor.getThreadCount(); i++) {
        producerChannels.add(clientConnection.createChannel());
    }
    // ensure the exchange is there
    consumerChannel.exchangeDeclare(exchangeName,"direct",true);
    if(ackType == BUFFERED) {
        messageAcker = new BufferingMessageAcker(consumerChannel);
    } else if(ackType == WRITE_BEHIND) {
        messageAcker = new WriteBehindMessageAcker(consumerChannel);
    } else if(ackType == ASYNC) {
        messageAcker = new AsyncMessageAcker(consumerChannel);
    } else {
        messageAcker = new DirectMessageAcker(consumerChannel);
    }
    messageAcker.start();
}