natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support offline notify (#12211) 

* Support offline notify

* Add log
This commit is contained in:
Albumen Kevin 2023-05-02 21:25:11 +08:00 committed by GitHub
parent 830c460c0a
commit 25a08a2b40
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 178 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -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";

8
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java
View File

@ -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<GracefulShutdown> 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()) {

47
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/GracefulShutdown.java Normal file
View File

@ -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";
}
}

1
dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.qos.api.BaseCommand
View File

@ -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

2
dubbo-plugin/dubbo-qos/src/test/java/org/apache/dubbo/qos/command/util/CommandHelperTest.java
View File

@ -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])));
}

6
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java
View File

@ -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);
}
}

31
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/GracefulShutdown.java Normal file
View File

@ -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<GracefulShutdown> getGracefulShutdowns(FrameworkModel frameworkModel) {
return frameworkModel.getBeanFactory().getBeansOfType(GracefulShutdown.class);
}
}

81
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboGracefulShutdown.java Normal file
View File

@ -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<Channel> 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);
}
}
}

1
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java
View File

@ -235,6 +235,7 @@ public class DubboProtocol extends AbstractProtocol {
}
};
this.frameworkModel = frameworkModel;
this.frameworkModel.getBeanFactory().registerBean(new DubboGracefulShutdown(this));
}
/**