Support multiple shared links

dubbo-common/src/main/java/org/apache/dubbo/common/Constants.java

@@ -143,7 +143,13 @@ public class Constants {
 
     public static final int DEFAULT_ALIVE = 60 * 1000;
 
-    public static final int DEFAULT_CONNECTIONS = 0;
+    /**
+     * By default, a consumer JVM instance and a provider JVM instance share a long TCP connection (except when connections are set),
+     * which can set the number of long TCP connections shared to avoid the bottleneck of sharing a single long TCP connection.
+     */
+    public static final String DEFAULT_SHARE_CONNECTIONS = "1";
+
+    public static final String SHARE_CONNECTIONS_KEY = "shareconnections";
 
     public static final int DEFAULT_ACCEPTS = 0;
 

dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ConsumerConfig.java

@@ -57,6 +57,12 @@ public class ConsumerConfig extends AbstractReferenceConfig {
      */
     private Integer queues;
 
+    /**
+     * By default, a TCP long-connection communication is shared between the consumer process and the provider process.
+     * This property can be set to share multiple TCP long-connection communications. Note that only the dubbo protocol takes effect.
+     */
+    private Integer shareconnections;
+
     @Override
     public void setTimeout(Integer timeout) {
         super.setTimeout(timeout);
@@ -118,4 +124,12 @@ public Integer getQueues() {
     public void setQueues(Integer queues) {
         this.queues = queues;
     }
+
+    public Integer getShareconnections() {
+        return shareconnections;
+    }
+
+    public void setShareconnections(Integer shareconnections) {
+        this.shareconnections = shareconnections;
+    }
 }

dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd

@@ -891,6 +891,12 @@
                         <xsd:documentation><![CDATA[ The thread pool queue size. ]]></xsd:documentation>
                     </xsd:annotation>
                 </xsd:attribute>
+                <xsd:attribute name="shareconnections" type="xsd:string">
+                    <xsd:annotation>
+                        <xsd:documentation>
+                            <![CDATA[ The default share connections. default share one connection. ]]></xsd:documentation>
+                    </xsd:annotation>
+                </xsd:attribute>
                 <xsd:anyAttribute namespace="##other" processContents="lax"/>
             </xsd:extension>
         </xsd:complexContent>

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractClient.java

@@ -174,30 +174,38 @@ public void send(Object message, boolean sent) throws RemotingException {
     }
 
     protected void connect() throws RemotingException {
+
         connectLock.lock();
+
         try {
+
             if (isConnected()) {
                 return;
             }
 
             doConnect();
+
             if (!isConnected()) {
                 throw new RemotingException(this, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
                         + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
                         + ", cause: Connect wait timeout: " + getConnectTimeout() + "ms.");
+
             } else {
                 if (logger.isInfoEnabled()) {
                     logger.info("Successed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
                             + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
                             + ", channel is " + this.getChannel());
                 }
             }
+
         } catch (RemotingException e) {
             throw e;
+
         } catch (Throwable e) {
             throw new RemotingException(this, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
                     + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
                     + ", cause: " + e.getMessage(), e);
+
         } finally {
             connectLock.unlock();
         }
@@ -241,23 +249,27 @@ public void reconnect() throws RemotingException {
 
     @Override
     public void close() {
+
         try {
             super.close();
         } catch (Throwable e) {
             logger.warn(e.getMessage(), e);
         }
+
         try {
             if (executor != null) {
                 ExecutorUtil.shutdownNow(executor, 100);
             }
         } catch (Throwable e) {
             logger.warn(e.getMessage(), e);
         }
+
         try {
             disconnect();
         } catch (Throwable e) {
             logger.warn(e.getMessage(), e);
         }
+
         try {
             doClose();
         } catch (Throwable e) {
@@ -310,5 +322,4 @@ public String toString() {
      * @return channel
      */
     protected abstract Channel getChannel();
-
 }