add RetryableConnection for the case for wait a moment when The chann…

…elMax limit is reached. Try later.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java

@@ -102,6 +102,7 @@ public enum AddressShuffleMode {
 
 	public static final int DEFAULT_CLOSE_TIMEOUT = 30000;
 
+
 	private static final String BAD_URI = "setUri() was passed an invalid URI; it is ignored";
 
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
@@ -159,6 +160,10 @@ public void handleRecovery(Recoverable recoverable) {
 
 	private volatile boolean contextStopped;
 
+	private int channelCreateTimeOut;
+
+	private int channelCreateRetryTimes;
+
 	/**
 	 * Create a new AbstractConnectionFactory for the given target ConnectionFactory,
 	 * with no publisher connection factory.
@@ -256,6 +261,14 @@ public void setHost(String host) {
 		this.rabbitConnectionFactory.setHost(host);
 	}
 
+	public void setChannelCreateTimeOut(int channelCreateTimeOut) {
+		this.channelCreateTimeOut = channelCreateTimeOut;
+	}
+
+	public void setChannelCreateRetryTimes(int channelCreateRetryTimes) {
+		this.channelCreateRetryTimes = channelCreateRetryTimes;
+	}
+
 	/**
 	 * Set the {@link ThreadFactory} on the underlying rabbit connection factory.
 	 * @param threadFactory the thread factory.
@@ -555,8 +568,14 @@ protected final Connection createBareConnection() {
 			String connectionName = this.connectionNameStrategy.obtainNewConnectionName(this);
 
 			com.rabbitmq.client.Connection rabbitConnection = connect(connectionName);
-
-			Connection connection = new SimpleConnection(rabbitConnection, this.closeTimeout);
+			Connection connection;
+			if (this.channelCreateRetryTimes > 0 && this.channelCreateTimeOut > 0) {
+				connection = new RetryableConnection(rabbitConnection, this.closeTimeout, this.channelCreateTimeOut,
+						this.channelCreateRetryTimes);
+			}
+			else {
+				connection = new SimpleConnection(rabbitConnection, this.closeTimeout);
+			}
 			if (rabbitConnection instanceof AutorecoveringConnection auto) {
 				auto.addRecoveryListener(new RecoveryListener() {
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RetryableConnection.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.connection;
+
+import java.io.IOException;
+
+import org.springframework.amqp.AmqpResourceNotAvailableException;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+/**
+ *
+ * extend from {@link SimpleConnection} provide retry ability
+ * @author salkli
+ * @date 2023/11/13
+ */
+public class RetryableConnection extends SimpleConnection {
+
+	private final com.rabbitmq.client.Connection delegate;
+
+	private int createTimeOut;
+
+	private int createTryTimes;
+
+	public RetryableConnection(Connection delegate, int closeTimeout, int createTimeOut, int createTryTimes) {
+		super(delegate, closeTimeout);
+		Assert.isTrue(createTimeOut >= 1, "channel create wait timeout must be 1 or higher");
+		Assert.isTrue(createTryTimes >= 1, "channel create retry times must be 1 or higher");
+		this.delegate = delegate;
+		this.createTimeOut = createTimeOut;
+		this.createTryTimes = createTryTimes;
+	}
+
+	@Override
+	public Channel createChannel(boolean transactional) {
+		try {
+			Channel channel = null;
+			for (int i = 0; i < this.createTryTimes; i++) {
+				channel = this.delegate.createChannel();
+				if (this.createTimeOut > 0) {
+					try {
+						Thread.sleep(createTimeOut * 1000);
+					}
+					catch (InterruptedException e) {
+						throw new RuntimeException(e);
+					}
+				}
+				if (channel != null) {
+					break;
+				}
+			}
+			if (channel == null) {
+				throw new AmqpResourceNotAvailableException("The channelMax limit is reached. Try later.");
+			}
+			if (transactional) {
+				// Just created so we want to start the transaction
+				channel.txSelect();
+			}
+			return channel;
+		}
+		catch (IOException e) {
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+		}
+	}
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java

@@ -1966,5 +1966,32 @@ public void onShutDown(ShutdownSignalException signal) {
 		assertThat(connShutDown.get()).isTrue();
 		assertThat(chanShutDown.get()).isFalse();
 	}
+	@Test
+	public void testRetryParam(){
+		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
+		ccf.setChannelCreateTimeOut(5);
+		ccf.setChannelCreateRetryTimes(3);
+		Connection bareConnection = ccf.createBareConnection();
+		assertThat(bareConnection.getClass()).isEqualTo(RetryableConnection.class);
+	}
+
+
+	@Test
+	public void testWaitForCreateChannel() throws Exception {
+		int createTimeOut = 3;
+		int createTryTimes = 2;
+		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
+		given(mockConnection.createChannel()).willReturn(null);
+		RetryableConnection retryableConnection = new RetryableConnection(mockConnection, 5, createTimeOut,
+				createTryTimes);
+		long time1 = System.currentTimeMillis();
+		try {
+			retryableConnection.createChannel(false);
+		} catch (Exception e) {
+		}
+		long time2 = System.currentTimeMillis();
+		assertThat((time2 - time1) / 1000).isGreaterThanOrEqualTo(createTimeOut * createTryTimes);
+	}
 
 }