Use multiple separate counters for the single service with multiple protocols.

extension-impl/extension-common/src/main/java/com/alipay/sofa/rpc/bootstrap/DefaultProviderBootstrap.java

@@ -110,36 +110,47 @@ private void doExport() {
         if (exported) {
             return;
         }
-        String key = providerConfig.buildKey();
-        String appName = providerConfig.getAppName();
+
         // 检查参数
         checkParameters();
-        if (LOGGER.isInfoEnabled(appName)) {
-            LOGGER.infoWithApp(appName, "Export provider config : {} with bean id {}", key, providerConfig.getId());
-        }
 
-        // 注意同一interface，同一uniqleId，不同server情况
-        AtomicInteger cnt = EXPORTED_KEYS.get(key); // 计数器
-        if (cnt == null) { // 没有发布过
-            cnt = CommonUtils.putToConcurrentMap(EXPORTED_KEYS, key, new AtomicInteger(0));
-        }
-        int c = cnt.incrementAndGet();
-        int maxProxyCount = providerConfig.getRepeatedExportLimit();
-        if (maxProxyCount > 0) {
-            if (c > maxProxyCount) {
-                cnt.decrementAndGet();
-                // 超过最大数量，直接抛出异常
-                throw new SofaRpcRuntimeException("Duplicate provider config with key " + key
-                    + " has been exported more than " + maxProxyCount + " times!"
-                    + " Maybe it's wrong config, please check it."
-                    + " Ignore this if you did that on purpose!");
-            } else if (c > 1) {
-                if (LOGGER.isInfoEnabled(appName)) {
-                    LOGGER.infoWithApp(appName, "Duplicate provider config with key {} has been exported!"
+        String appName = providerConfig.getAppName();
+
+        // 将处理器注册到server
+        List<ServerConfig> serverConfigs = providerConfig.getServer();
+        for (ServerConfig serverConfig : serverConfigs) {
+            String protocol = serverConfig.getProtocol();
+
+            String key = providerConfig.buildKey() + ":" + protocol;
+
+            if (LOGGER.isInfoEnabled(appName)) {
+                LOGGER.infoWithApp(appName, "Export provider config : {} with bean id {}", key, providerConfig.getId());
+            }
+
+            // 注意同一interface，同一uniqleId，不同server情况
+            AtomicInteger cnt = EXPORTED_KEYS.get(key); // 计数器
+            if (cnt == null) { // 没有发布过
+                cnt = CommonUtils.putToConcurrentMap(EXPORTED_KEYS, key, new AtomicInteger(0));
+            }
+            int c = cnt.incrementAndGet();
+            int maxProxyCount = providerConfig.getRepeatedExportLimit();
+            if (maxProxyCount > 0) {
+                if (c > maxProxyCount) {
+                    cnt.decrementAndGet();
+                    // 超过最大数量，直接抛出异常
+                    throw new SofaRpcRuntimeException("Duplicate provider config with key " + key
+                        + " has been exported more than " + maxProxyCount + " times!"
                         + " Maybe it's wrong config, please check it."
-                        + " Ignore this if you did that on purpose!", key);
+                        + " Ignore this if you did that on purpose!");
+                } else if (c > 1) {
+                    if (LOGGER.isInfoEnabled(appName)) {
+                        LOGGER.infoWithApp(appName, "Duplicate provider config with key {} has been exported!"
+                            + " Maybe it's wrong config, please check it."
+                            + " Ignore this if you did that on purpose!", key);
+                    }
                 }
             }
+
         }
 
         try {
@@ -155,7 +166,6 @@ private void doExport() {
                 }
             }
             // 将处理器注册到server
-            List<ServerConfig> serverConfigs = providerConfig.getServer();
             for (ServerConfig serverConfig : serverConfigs) {
                 try {
                     Server server = serverConfig.buildIfAbsent();
@@ -176,7 +186,17 @@ private void doExport() {
             providerConfig.setConfigListener(new ProviderAttributeListener());
             register();
         } catch (Exception e) {
-            cnt.decrementAndGet();
+
+            //once error, we decrementAndGet all counter
+            for (ServerConfig serverConfig : serverConfigs) {
+                String protocol = serverConfig.getProtocol();
+                String key = providerConfig.buildKey() + ":" + protocol;
+                AtomicInteger cnt = EXPORTED_KEYS.get(key); // 计数器
+                if (cnt != null && cnt.get() > 0) {
+                    cnt.decrementAndGet();
+                }
+            }
+
             if (e instanceof SofaRpcRuntimeException) {
                 throw (SofaRpcRuntimeException) e;
             } else {
@@ -246,12 +266,16 @@ public void unExport() {
             if (!exported) {
                 return;
             }
-
-            String key = providerConfig.buildKey();
             String appName = providerConfig.getAppName();
-            if (LOGGER.isInfoEnabled(appName)) {
-                LOGGER.infoWithApp(appName, "Unexport provider config : {} {}", key, providerConfig.getId() != null
-                    ? "with bean id " + providerConfig.getId() : "");
+
+            List<ServerConfig> serverConfigs = providerConfig.getServer();
+            for (ServerConfig serverConfig : serverConfigs) {
+                String protocol = serverConfig.getProtocol();
+                String key = providerConfig.buildKey() + ":" + protocol;
+                if (LOGGER.isInfoEnabled(appName)) {
+                    LOGGER.infoWithApp(appName, "Unexport provider config : {} {}", key, providerConfig.getId() != null
+                        ? "with bean id " + providerConfig.getId() : "");
+                }
             }
 
             // 取消注册到注册中心
@@ -260,7 +284,6 @@ public void unExport() {
             providerProxyInvoker = null;
 
             // 取消将处理器注册到server
-            List<ServerConfig> serverConfigs = providerConfig.getServer();
             if (serverConfigs != null) {
                 for (ServerConfig serverConfig : serverConfigs) {
                     Server server = serverConfig.getServer();
@@ -281,10 +304,15 @@ public void unExport() {
             providerConfig.setConfigListener(null);
 
             // 清除缓存状态
-            AtomicInteger cnt = EXPORTED_KEYS.get(key);
-            if (cnt != null && cnt.decrementAndGet() <= 0) {
-                EXPORTED_KEYS.remove(key);
+            for (ServerConfig serverConfig : serverConfigs) {
+                String protocol = serverConfig.getProtocol();
+                String key = providerConfig.buildKey() + ":" + protocol;
+                AtomicInteger cnt = EXPORTED_KEYS.get(key);
+                if (cnt != null && cnt.decrementAndGet() <= 0) {
+                    EXPORTED_KEYS.remove(key);
+                }
             }
+
             RpcRuntimeContext.invalidateProviderConfig(this);
             exported = false;
         }

test/test-integration/src/test/java/com/alipay/sofa/rpc/server/multi/MultiProtolServerTest.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *     http://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 com.alipay.sofa.rpc.server.multi;
+
+import com.alipay.sofa.rpc.common.RpcConstants;
+import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.ServerConfig;
+import com.alipay.sofa.rpc.context.RpcRuntimeContext;
+import com.alipay.sofa.rpc.server.rest.RestService;
+import com.alipay.sofa.rpc.server.rest.RestServiceImpl;
+import com.alipay.sofa.rpc.test.ActivelyDestroyTest;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author <a href="mailto:zhanggeng.zg@antfin.com">GengZhang</a>
+ */
+public class MultiProtolServerTest extends ActivelyDestroyTest {
+
+    @Test
+    public void testAll() {
+        testMultiProtol();
+        RpcRuntimeContext.destroy();
+    }
+
+    public void testMultiProtol() {
+
+        try {
+            // 只有2个线程 执行
+            ServerConfig serverConfig = new ServerConfig()
+                .setStopTimeout(0)
+                .setPort(22222)
+                .setProtocol(RpcConstants.PROTOCOL_TYPE_BOLT)
+                .setQueues(100).setCoreThreads(1).setMaxThreads(2);
+
+            // 发布一个服务，每个请求要执行1秒
+            ProviderConfig<RestService> providerConfig = new ProviderConfig<RestService>()
+                .setInterfaceId(RestService.class.getName())
+                .setRef(new RestServiceImpl())
+                .setServer(serverConfig)
+                .setRepeatedExportLimit(1)
+                .setRegister(false);
+            providerConfig.export();
+
+            ServerConfig serverConfig2 = new ServerConfig()
+                .setStopTimeout(0)
+                .setPort(22223)
+                .setProtocol(RpcConstants.PROTOCOL_TYPE_REST)
+                .setQueues(100).setCoreThreads(1).setMaxThreads(2);
+
+            // 发布一个服务，每个请求要执行1秒
+            ProviderConfig<RestService> providerConfig2 = new ProviderConfig<RestService>()
+                .setInterfaceId(RestService.class.getName())
+                .setRef(new RestServiceImpl())
+                .setServer(serverConfig2)
+                .setRepeatedExportLimit(1)
+                .setRegister(false);
+            providerConfig2.export();
+        } catch (Throwable e) {
+            Assert.fail();
+        }
+
+    }
+}