diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index f066044043..978d053031 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -313,6 +313,7 @@ public interface CommonConstants { String HEARTBEAT_EVENT = null; String MOCK_HEARTBEAT_EVENT = "H"; String READONLY_EVENT = "R"; + String WRITEABLE_EVENT = "W"; String REFERENCE_FILTER_KEY = "reference.filter"; diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java index c946fa4390..3172525e72 100644 --- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java +++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java @@ -22,9 +22,11 @@ import org.apache.dubbo.common.constants.LoggerCodeConstants; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.Assert; +import org.apache.dubbo.rpc.GracefulShutdown; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ModuleModel; +import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -80,6 +82,12 @@ public class DubboShutdownHook extends Thread { } private void doDestroy() { + // send readonly for shutdown hook + List gracefulShutdowns = GracefulShutdown.getGracefulShutdowns(applicationModel.getFrameworkModel()); + for (GracefulShutdown gracefulShutdown : gracefulShutdowns) { + gracefulShutdown.readonly(); + } + boolean hasModuleBindSpring = false; // check if any modules are bound to Spring for (ModuleModel module: applicationModel.getModuleModels()) { diff --git a/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/GracefulShutdown.java b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/GracefulShutdown.java new file mode 100644 index 0000000000..c4fa69557d --- /dev/null +++ b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/GracefulShutdown.java @@ -0,0 +1,47 @@ +/* + * 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 org.apache.dubbo.qos.command.impl; + +import org.apache.dubbo.qos.api.BaseCommand; +import org.apache.dubbo.qos.api.Cmd; +import org.apache.dubbo.qos.api.CommandContext; +import org.apache.dubbo.qos.api.PermissionLevel; +import org.apache.dubbo.rpc.model.FrameworkModel; + +@Cmd(name = "gracefulShutdown", + summary = "Gracefully shutdown servers", + example = {"gracefulShutdown"}, + requiredPermissionLevel = PermissionLevel.PRIVATE) +public class GracefulShutdown implements BaseCommand { + private final Offline offline; + private final FrameworkModel frameworkModel; + + public GracefulShutdown(FrameworkModel frameworkModel) { + this.offline = new Offline(frameworkModel); + this.frameworkModel = frameworkModel; + } + + @Override + public String execute(CommandContext commandContext, String[] args) { + offline.execute(commandContext, new String[0]); + for (org.apache.dubbo.rpc.GracefulShutdown gracefulShutdown : + org.apache.dubbo.rpc.GracefulShutdown.getGracefulShutdowns(frameworkModel)) { + gracefulShutdown.readonly(); + } + return "OK"; + } +} diff --git a/dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.qos.api.BaseCommand b/dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.qos.api.BaseCommand index 78825644ed..d575733b46 100644 --- a/dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.qos.api.BaseCommand +++ b/dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.qos.api.BaseCommand @@ -36,3 +36,4 @@ serializeCheckStatus=org.apache.dubbo.qos.command.impl.SerializeCheckStatus serializeWarnedClasses=org.apache.dubbo.qos.command.impl.SerializeWarnedClasses getConfig=org.apache.dubbo.qos.command.impl.GetConfig getAddress=org.apache.dubbo.qos.command.impl.GetAddress +gracefulShutdown=org.apache.dubbo.qos.command.impl.GracefulShutdown diff --git a/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/command/util/CommandHelperTest.java b/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/command/util/CommandHelperTest.java index a06a76408c..0370e519ef 100644 --- a/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/command/util/CommandHelperTest.java +++ b/dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/command/util/CommandHelperTest.java @@ -30,6 +30,7 @@ import org.apache.dubbo.qos.command.impl.GetConfig; import org.apache.dubbo.qos.command.impl.GetEnabledRouterSnapshot; import org.apache.dubbo.qos.command.impl.GetRecentRouterSnapshot; import org.apache.dubbo.qos.command.impl.GetRouterSnapshot; +import org.apache.dubbo.qos.command.impl.GracefulShutdown; import org.apache.dubbo.qos.command.impl.Help; import org.apache.dubbo.qos.command.impl.InvokeTelnet; import org.apache.dubbo.qos.command.impl.Live; @@ -123,6 +124,7 @@ class CommandHelperTest { expectedClasses.add(SerializeWarnedClasses.class); expectedClasses.add(GetConfig.class); expectedClasses.add(GetAddress.class); + expectedClasses.add(GracefulShutdown.class); assertThat(classes, containsInAnyOrder(expectedClasses.toArray(new Class[0]))); } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java index ef0420f8f4..fd9d660d05 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java @@ -38,6 +38,7 @@ import java.net.InetSocketAddress; import java.util.concurrent.CompletionStage; import static org.apache.dubbo.common.constants.CommonConstants.READONLY_EVENT; +import static org.apache.dubbo.common.constants.CommonConstants.WRITEABLE_EVENT; import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RESPONSE; import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_UNSUPPORTED_MESSAGE; @@ -75,6 +76,11 @@ public class HeaderExchangeHandler implements ChannelHandlerDelegate { void handlerEvent(Channel channel, Request req) throws RemotingException { if (req.getData() != null && req.getData().equals(READONLY_EVENT)) { channel.setAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY, Boolean.TRUE); + logger.info("ChannelReadOnly set true for channel: " + channel); + } + if (req.getData() != null && req.getData().equals(WRITEABLE_EVENT)) { + channel.removeAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY); + logger.info("ChannelReadOnly set false for channel: " + channel); } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/GracefulShutdown.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/GracefulShutdown.java new file mode 100644 index 0000000000..e8bdb38044 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/GracefulShutdown.java @@ -0,0 +1,31 @@ +/* + * 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 org.apache.dubbo.rpc; + +import org.apache.dubbo.rpc.model.FrameworkModel; + +import java.util.List; + +public interface GracefulShutdown { + void readonly(); + + void writeable(); + + static List getGracefulShutdowns(FrameworkModel frameworkModel) { + return frameworkModel.getBeanFactory().getBeansOfType(GracefulShutdown.class); + } +} diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboGracefulShutdown.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboGracefulShutdown.java new file mode 100644 index 0000000000..f1a324e01a --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboGracefulShutdown.java @@ -0,0 +1,81 @@ +/* + * 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 org.apache.dubbo.rpc.protocol.dubbo; + +import org.apache.dubbo.common.Version; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.remoting.Channel; +import org.apache.dubbo.remoting.Constants; +import org.apache.dubbo.remoting.RemotingException; +import org.apache.dubbo.remoting.exchange.Request; +import org.apache.dubbo.rpc.GracefulShutdown; +import org.apache.dubbo.rpc.ProtocolServer; + +import java.nio.channels.ClosedChannelException; +import java.util.Collection; + +import static org.apache.dubbo.common.constants.CommonConstants.READONLY_EVENT; +import static org.apache.dubbo.common.constants.CommonConstants.WRITEABLE_EVENT; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE_STREAM; + +public class DubboGracefulShutdown implements GracefulShutdown { + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(DubboGracefulShutdown.class); + private final DubboProtocol dubboProtocol; + + public DubboGracefulShutdown(DubboProtocol dubboProtocol) { + this.dubboProtocol = dubboProtocol; + } + + @Override + public void readonly() { + sendEvent(READONLY_EVENT); + } + + @Override + public void writeable() { + sendEvent(WRITEABLE_EVENT); + } + + private void sendEvent(String event) { + try { + for (ProtocolServer server : dubboProtocol.getServers()) { + Collection channels = server.getRemotingServer().getChannels(); + Request request = new Request(); + request.setEvent(event); + request.setTwoWay(false); + request.setVersion(Version.getProtocolVersion()); + + for (Channel channel : channels) { + try { + if (channel.isConnected()) { + channel.send(request, channel.getUrl().getParameter(Constants.CHANNEL_READONLYEVENT_SENT_KEY, true)); + } + } catch (RemotingException e) { + if (e.getCause() instanceof ClosedChannelException) { + // ignore ClosedChannelException which means the connection has been closed. + continue; + } + logger.warn(TRANSPORT_FAILED_CLOSE_STREAM, "", "", "send cannot write message error.", e); + } + } + } + } catch (Throwable e) { + logger.warn(TRANSPORT_FAILED_CLOSE_STREAM, "", "", "send cannot write message error.", e); + } + } +} diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java index 0e10031e4a..3c0138e0d6 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java @@ -235,6 +235,7 @@ public class DubboProtocol extends AbstractProtocol { } }; this.frameworkModel = frameworkModel; + this.frameworkModel.getBeanFactory().registerBean(new DubboGracefulShutdown(this)); } /**