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

DUBBO-71 Graceful shutdown-半关闭状态 

git-svn-id: http://code.alibabatech.com/svn/dubbo/trunk@339 1a56cb94-b969-4eaa-88fa-be21384802f2
This commit is contained in:
chao.liuc 2011-11-16 07:55:05 +00:00
parent edc103a164
commit 9408683d48
9 changed files with 192 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java
View File

@ -277,9 +277,13 @@ public class Constants {
public static final String CONNECT_QUENE_WARNING_SIZE = "connect.quene.warning.size";
public static final int DEFAULT_CONNECT_QUENE_WARNING_SIZE = 1000;
public static final int DEFAULT_CONNECT_QUENE_WARNING_SIZE = 1000;
public static final String CHANNEL_CANNOTWRITE_KEY = "cannotwrite";
public static final String CHANNEL_ATTRIBUTE_READONLY_KEY = "channel.readonly";
public static final String CHANNEL_READONLYEVENT_SENT_KEY = "channel.readonly.sent";
public static final String CHANNEL_SEND_READONLYEVENT_KEY = "channel.readonly.send";
private Constants(){
}

44
dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/exchange/Request.java
View File

@ -23,17 +23,21 @@ import java.util.concurrent.atomic.AtomicLong;
* @author qian.lei
* @author william.liangf
*/
public class Request {
public class Request {
public static final String HEARTBEAT_EVENT = null;
public static final String READONLY_EVENT = "R";
private static final AtomicLong INVOKE_ID = new AtomicLong(0);
private final long mId;
private String mVersion;
private boolean mTwoWay = true;
private boolean mHeatbeat = false;
private boolean mTwoWay = true;
private boolean mEvent = false;
private boolean mBroken = false;
@ -66,14 +70,15 @@ public class Request {
public void setTwoWay(boolean twoWay) {
mTwoWay = twoWay;
}
public boolean isHeartbeat() {
return mHeatbeat;
}
public void setHeartbeat(boolean isHeartbeat) {
this.mHeatbeat = isHeartbeat;
}
public boolean isEvent() {
return mEvent;
}
public void setEvent(String event) {
mEvent = true;
mData = event;
}
public boolean isBroken() {
return mBroken;
@ -90,6 +95,17 @@ public class Request {
public void setData(Object msg) {
mData = msg;
}
public boolean isHeartbeat() {
return mEvent && HEARTBEAT_EVENT.equals(mData);
}
@Deprecated
public void setHeartbeat(boolean isHeartbeat) {
if (isHeartbeat) {
setEvent(HEARTBEAT_EVENT);
}
}
private static long newId() {
// getAndIncrement()增长到MAX_VALUE时，再增长会变为MIN_VALUE，负数也可以做为ID
@ -98,7 +114,7 @@ public class Request {
@Override
public String toString() {
return "Request [id=" + mId + ", version=" + mVersion + ", twoway=" + mTwoWay + ", heatbeat=" + mHeatbeat
return "Request [id=" + mId + ", version=" + mVersion + ", twoway=" + mTwoWay + ", event=" + mEvent
+ ", broken=" + mBroken + ", data=" + (mData == this ? "this" : mData) + "]";
}

33
dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/exchange/Response.java
View File

@ -21,7 +21,10 @@ package com.alibaba.dubbo.remoting.exchange;
* @author qian.lei
* @author william.liangf
*/
public class Response {
public class Response {
public static final String HEARTBEAT_EVENT = null;
public static final String READONLY_EVENT = "R";
/**
* ok.
@ -69,7 +72,7 @@ public class Response {
private byte mStatus = OK;
private boolean mHeatbeat = false;
private boolean mEvent = false;
private String mErrorMsg;
@ -109,14 +112,26 @@ public class Response {
public void setStatus(byte status) {
mStatus = status;
}
public boolean isEvent() {
return mEvent;
}
public void setEvent(String event) {
mEvent = true;
mResult = event;
}
public boolean isHeartbeat() {
return mHeatbeat;
}
public void setHeartbeat(boolean isHeatbeat) {
this.mHeatbeat = isHeatbeat;
public boolean isHeartbeat() {
return mEvent && HEARTBEAT_EVENT.equals(mResult);
}
@Deprecated
public void setHeartbeat(boolean isHeartbeat) {
if (isHeartbeat) {
setEvent(HEARTBEAT_EVENT);
}
}
public Object getResult() {
@ -137,7 +152,7 @@ public class Response {
@Override
public String toString() {
return "Response [id=" + mId + ", version=" + mVersion + ", status=" + mStatus + ", heatbeat=" + mHeatbeat
return "Response [id=" + mId + ", version=" + mVersion + ", status=" + mStatus + ", event=" + mEvent
+ ", error=" + mErrorMsg + ", result=" + (mResult == this ? "this" : mResult) + "]";
}
}

65
dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/exchange/codec/ExchangeCodec.java
View File

@ -65,7 +65,7 @@ public class ExchangeCodec extends TelnetCodec {
protected static final byte FLAG_TWOWAY = (byte) 0x40;
protected static final byte FLAG_HEARTBEAT = (byte) 0x20;
protected static final byte FLAG_EVENT = (byte) 0x20;
protected static final int SERIALIZATION_MASK = 0x1f;
@ -154,16 +154,20 @@ public class ExchangeCodec extends TelnetCodec {
long id = Bytes.bytes2long(header, 4);
if( ( flag & FLAG_REQUEST ) == 0 ) {
// decode response.
Response res = new Response(id);
res.setHeartbeat( ( flag & FLAG_HEARTBEAT ) != 0 );
Response res = new Response(id);
if (( flag & FLAG_EVENT ) != 0){
res.setEvent(Response.HEARTBEAT_EVENT);
}
// get status.
byte status = header[3];
res.setStatus(status);
if( status == Response.OK ) {
try {
Object data;
if (res.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
if (res.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
} else if (res.isEvent()) {
data = decodeEventData(channel, in);
} else {
data = decodeResponseData(channel, in);
}
@ -180,12 +184,16 @@ public class ExchangeCodec extends TelnetCodec {
// decode request.
Request req = new Request(id);
req.setVersion("2.0.0");
req.setTwoWay( ( flag & FLAG_TWOWAY ) != 0 );
req.setHeartbeat( ( flag & FLAG_HEARTBEAT ) != 0 );
req.setTwoWay( ( flag & FLAG_TWOWAY ) != 0 );
if (( flag & FLAG_EVENT ) != 0 ){
req.setEvent(Request.HEARTBEAT_EVENT);
}
try {
Object data;
if (req.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
if (req.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
} else if (req.isEvent()) {
data = decodeEventData(channel, in);
} else {
data = decodeRequestData(channel, in);
}
@ -210,7 +218,7 @@ public class ExchangeCodec extends TelnetCodec {
header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;
if (req.isHeartbeat()) header[2] |= FLAG_HEARTBEAT;
if (req.isEvent()) header[2] |= FLAG_EVENT;
// set request id.
Bytes.long2bytes(req.getId(), header, 4);
@ -218,8 +226,8 @@ public class ExchangeCodec extends TelnetCodec {
// encode request data.
UnsafeByteArrayOutputStream bos = new UnsafeByteArrayOutputStream(1024);
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
if (req.isHeartbeat()) {
encodeHeartbeatData(channel, out, req.getData());
if (req.isEvent()) {
encodeEventData(channel, out, req.getData());
} else {
encodeRequestData(channel, out, req.getData());
}
@ -242,7 +250,7 @@ public class ExchangeCodec extends TelnetCodec {
Bytes.short2bytes(MAGIC, header);
// set request and serialization flag.
header[2] = serialization.getContentTypeId();
if (res.isHeartbeat()) header[2] |= FLAG_HEARTBEAT;
if (res.isHeartbeat()) header[2] |= FLAG_EVENT;
// set response status.
byte status = res.getStatus();
header[3] = status;
@ -253,7 +261,7 @@ public class ExchangeCodec extends TelnetCodec {
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
// encode response data or error message.
if (status == Response.OK) {
if (res.isHeartbeat()) {
if (res.isHeartbeat()) {
encodeHeartbeatData(channel, out, res.getResult());
} else {
encodeResponseData(channel, out, res.getResult());
@ -279,7 +287,8 @@ public class ExchangeCodec extends TelnetCodec {
protected Object decodeData(ObjectInput in) throws IOException {
return decodeRequestData(in);
}
@Deprecated
protected Object decodeHeartbeatData(ObjectInput in) throws IOException {
try {
return in.readObject();
@ -308,9 +317,14 @@ public class ExchangeCodec extends TelnetCodec {
protected void encodeData(ObjectOutput out, Object data) throws IOException {
encodeRequestData(out, data);
}
private void encodeEventData(ObjectOutput out, Object data) throws IOException {
out.writeObject(data);
}
@Deprecated
protected void encodeHeartbeatData(ObjectOutput out, Object data) throws IOException {
out.writeObject(data);
encodeEventData(out, data);
}
protected void encodeRequestData(ObjectOutput out, Object data) throws IOException {
@ -324,8 +338,17 @@ public class ExchangeCodec extends TelnetCodec {
@Override
protected Object decodeData(Channel channel, ObjectInput in) throws IOException {
return decodeRequestData(channel ,in);
}
private Object decodeEventData(Channel channel, ObjectInput in) throws IOException {
try {
return in.readObject();
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read object failed.", e));
}
}
@Deprecated
protected Object decodeHeartbeatData(Channel channel, ObjectInput in) throws IOException {
try {
return in.readObject();
@ -346,7 +369,11 @@ public class ExchangeCodec extends TelnetCodec {
protected void encodeData(Channel channel, ObjectOutput out, Object data) throws IOException {
encodeRequestData(channel, out, data);
}
private void encodeEventData(Channel channel, ObjectOutput out, Object data) throws IOException {
encodeEventData(out, data);
}
@Deprecated
protected void encodeHeartbeatData(Channel channel, ObjectOutput out, Object data) throws IOException {
encodeHeartbeatData(out, data);
}

37
dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java
View File

@ -15,6 +15,7 @@
*/
package com.alibaba.dubbo.remoting.exchange.support.header;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.StringUtils;
@ -49,14 +50,22 @@ public class HeaderExchangeHandler implements ChannelHandler {
}
this.handler = handler;
}
void handlerEvent(Channel channel, Request req) throws RemotingException{
if (req.getData() != null && req.getData().equals(Request.READONLY_EVENT)){
channel.setAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY, Boolean.TRUE);
}
if (req.isTwoWay()){
if (req.isHeartbeat()) {
Response res = new Response(req.getId(), req.getVersion());
res.setEvent(req.getData() == null ? null : req.getData().toString());
channel.send(res);
}
}
}
Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
Response res = new Response(req.getId(), req.getVersion());
if (req.isHeartbeat()) {
res.setHeartbeat(true);
return res;
}
if (req.isBroken()) {
Object data = req.getData();
@ -153,14 +162,18 @@ public class HeaderExchangeHandler implements ChannelHandler {
if (message instanceof Request) {
// handle request.
Request request = (Request) message;
if (request.isTwoWay()) {
Response response = handleRequest(exchangeChannel, request);
if (response == null) {
throw new RemotingException(channel, "Response is null.");
}
channel.send(response);
if (request.isEvent()){
handlerEvent(channel, request);
} else {
handler.received(exchangeChannel, request.getData());
if (request.isTwoWay()) {
Response response = handleRequest(exchangeChannel, request);
if (response == null) {
throw new RemotingException(channel, "Response is null.");
}
channel.send(response);
} else {
handler.received(exchangeChannel, request.getData());
}
}
} else if (message instanceof Response) {
handleResponse(channel, (Response) message);

22
dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/exchange/support/header/HeaderExchangeServer.java
View File

@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.Version;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.NamedThreadFactory;
@ -103,6 +104,9 @@ public class HeaderExchangeServer implements ExchangeServer {
if (timeout > 0) {
final long max = (long) timeout;
final long start = System.currentTimeMillis();
if (getUrl().getParameter(Constants.CHANNEL_SEND_READONLYEVENT_KEY, false)){
sendChannelReadOnlyEvent();
}
while (HeaderExchangeServer.this.isRunning()
&& System.currentTimeMillis() - start < max) {
try {
@ -116,6 +120,22 @@ public class HeaderExchangeServer implements ExchangeServer {
server.close(timeout);
}
private void sendChannelReadOnlyEvent(){
Request request = new Request();
request.setEvent(Request.READONLY_EVENT);
request.setTwoWay(false);
request.setVersion(Version.getVersion());
Collection<Channel> channels = getChannels();
for (Channel channel : channels) {
try {
if (channel.isConnected())channel.send(request, getUrl().getParameter(Constants.CHANNEL_READONLYEVENT_SENT_KEY, true));
} catch (RemotingException e) {
logger.warn("send connot write messge error.", e);
}
}
}
private void doClose() {
if (closed) {
return;
@ -245,7 +265,7 @@ public class HeaderExchangeServer implements ExchangeServer {
Request req = new Request();
req.setVersion("2.0.0");
req.setTwoWay(true);
req.setHeartbeat(true);
req.setEvent(Request.HEARTBEAT_EVENT);
channel.send(req);
if (logger.isDebugEnabled()) {
logger.debug("Send heartbeat to client " + channel.getRemoteAddress() + ".");

10
dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/transport/AbstractServer.java
View File

@ -137,11 +137,13 @@ public abstract class AbstractServer extends AbstractEndpoint implements Server
public void send(Object message, boolean sent) throws RemotingException {
Collection<Channel> channels = getChannels();
for (Channel channel : channels) {
channel.send(message, sent);
if (channel.isConnected()) {
channel.send(message, sent);
}
}
}
public void close() {
}
public void close() {
ExecutorUtil.shutdownNow(executor ,100);
try {
super.close();

21
dubbo-rpc-default/src/main/java/com/alibaba/dubbo/rpc/protocol/dubbo/DubboInvoker.java
View File

@ -104,12 +104,21 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
if (!super.isAvailable())
return false;
for (ExchangeClient client : clients){
//cannot write == not Available ?
// if (client.isConnected() && !client.hasAttribute(Constants.CHANNEL_CANNOTWRITE_KEY)){
// return true;
// }
if (client.isConnected() ){
return true;
if (client.isConnected()){
boolean isLazy = client.getUrl().getParameter(RpcConstants.LAZY_CONNECT_KEY, false);
//cannot write == not Available ?
if (! isLazy) {
return true;
} else if (client instanceof LazyConnectExchangeClient) {
LazyConnectExchangeClient lazyClient = (LazyConnectExchangeClient) client;
if (lazyClient.isInited() && lazyClient.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY)){
return false;
} else {
return true;
}
} else {
return true;
}
}
}
return false;

34
dubbo-rpc-default/src/main/java/com/alibaba/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java
View File

@ -15,22 +15,22 @@
*/
package com.alibaba.dubbo.rpc.protocol.dubbo;
import java.net.InetSocketAddress;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.alibaba.dubbo.common.Constants;
import java.net.InetSocketAddress;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.Parameters;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.remoting.ChannelHandler;
import com.alibaba.dubbo.remoting.RemotingException;
import com.alibaba.dubbo.remoting.exchange.ExchangeClient;
import com.alibaba.dubbo.remoting.exchange.ExchangeHandler;
import com.alibaba.dubbo.remoting.exchange.Exchangers;
import com.alibaba.dubbo.remoting.exchange.ResponseFuture;
import com.alibaba.dubbo.rpc.RpcConstants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.remoting.ChannelHandler;
import com.alibaba.dubbo.remoting.RemotingException;
import com.alibaba.dubbo.remoting.exchange.ExchangeClient;
import com.alibaba.dubbo.remoting.exchange.ExchangeHandler;
import com.alibaba.dubbo.remoting.exchange.Exchangers;
import com.alibaba.dubbo.remoting.exchange.ResponseFuture;
import com.alibaba.dubbo.rpc.RpcConstants;
/**
* dubbo protocol support class.
@ -179,5 +179,9 @@ final class LazyConnectExchangeClient implements ExchangeClient {
throw new IllegalStateException(
"LazyConnectExchangeClient state error. the client has not be init .url:" + url);
}
}
public boolean isInited(){
return client == null ;
}
}