Issue 1:
I have searched the issues of this repository and believe that this is not a duplicate.

I have checked the FAQ of this repository and believe that this is not a duplicate.

Environment

Dubbo version: 2.7.6

Operating System version: win10 x64

Java version: 1.8

Steps to reproduce this issue

dubbo从2.8.4 的版本升级到2.7.6的版本，生产者启动正常，消费者调用多个应用的生产实例化dubbo的bean时抛出异常。 低版本启动正常，使用正常。

Pls. provide [GitHub address] to reproduce this issue.

Expected Result

What do you expected from the above steps？

Actual Result

What actually happens?

If there is an exception, please attach the exception trace:

堆栈异常如下：
[stack trace.txt](https://github.com/apache/dubbo/files/4786508/stack.trace.txt)

Issue 2:
I have searched the issues of this repository and believe that this is not a duplicate.

I have checked the FAQ of this repository and believe that this is not a duplicate.

Environment

Dubbo version: 2.7.6

Operating System version: linux/windows

Java version: 1.8

简述一下问题

启动一个consumer和一个provider

在consumer中重复调用一个provider n次(执行时间长一点容易发现)

这时向consumer发送kill -15 命令

可能会报错:

org.apache.dubbo.rpc.RpcException: Failed to invoke the method sayHello in the service com.dubbo.springboot.demo.provider.service.DemoService. Tried 3 times of the providers [x.x.x.x:xxxx] (1/1) from the registry localhost:2181 on the consumer x.x.x.x using the dubbo version 2.7.6. Last error is: Failed to invoke remote method: sayHello, provider......

导致程序提前中断

Steps to reproduce this issue

provider service

`

package org.apache.dubbo.spring.boot.sample.provider.service;

import org.apache.dubbo.config.annotation.Service;

import org.springframework.beans.factory.annotation.Value;

import com.dubbo.springboot.demo.provider.service.DemoService;

/**

Default {@link DemoService}

@see DemoService

@SInCE 2.7.0

*/

@service(version = "${demo.service.version}", timeout = 30000)

public class DefaultDemoService implements DemoService {

/**

The default value of ${dubbo.application.name} is ${spring.application.name}

*/

@value("${dubbo.application.name}")

private String serviceName;

// @resource

// private KafkaTemplate<String, Object> kafkaTemplate;

@OverRide

public String sayHello(String name) {

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("sayHello");

return String.format("[%s] : Hello, %s", serviceName, name);

}

}

`

consumer service

`

package com.dubbo.springboot.demo.consumer.service;

import org.apache.dubbo.config.annotation.Reference;

import org.springframework.stereotype.Service;

import com.dubbo.springboot.demo.provider.service.DemoService;

@service

public class DemoConsumerService {

@Reference(version = "${demo.service.version}")
private DemoService demoService;

public String sayHello(String name) {
for (int i = 0; i < 10; i++) {
try {
demoService.sayHello(name);
} catch (Exception e) {
e.printStackTrace();
}
}

return "nb";
}

}

`

cause

`

org.apache.dubbo.remoting.exchange.support.DefaultFuture ==> 184行

private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<>();

public static void received(Channel channel, Response response, boolean timeout) {

try {

DefaultFuture future = FUTURES.remove(response.getId());

if (future != null) {

Timeout t = future.timeoutCheckTask;

if (!timeout) {

// decrease Time

t.cancel();

}

future.doReceived(response);

} else {

logger.warn("The timeout response finally returned at "

+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))

+ ", response " + response

+ (channel == null ? "" : ", channel: " + channel.getLocalAddress()

+ " -> " + channel.getRemoteAddress()));

}

} finally {

CHANNELS.remove(response.getId());// <== 这里会把channel删掉

}

}

org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeChannel ==> 168行

private volatile boolean closed = false;

public void close(int timeout) {

if (closed) {

return;

}

closed = true; // <== 这里会导致在下面循环进行时,其他请求进不来,

if (timeout > 0) {

long start = System.currentTimeMillis();

while (DefaultFuture.hasFuture(channel) // <== 因为1的原因这里会返回false

&& System.currentTimeMillis() - start < timeout) {

try {

Thread.sleep(10); // <==可能存在10ms,channle已经从1中的CHANNELS里remove掉了,但是没有被close掉,导致下一次请求时closed还是true,而不是新创建一个channel,导致4出现问题

} catch (InterruptedException e) {

logger.warn(e.getMessage(), e);

}

}

}

close();

}

org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeChannel ==> 93行

public void send(Object message, boolean sent) throws RemotingException {

if (closed) {

throw new RemotingException(this.getLocalAddress(), null, "Failed to send message " + message + ", cause: The channel " + this + " is closed!"); // <== 这里会因为2的原因报错

}

if (message instanceof Request

|| message instanceof Response

|| message instanceof String) {

channel.send(message, sent);

} else {

Request request = new Request();

request.setVersion(Version.getProtocolVersion());

request.setTwoWay(false);

request.setData(message);

channel.send(request, sent);

}

}

org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeChannel ==> 63行

static HeaderExchangeChannel getOrAddChannel(Channel ch) {

if (ch == null) {

return null;

}

HeaderExchangeChannel ret = (HeaderExchangeChannel) ch.getAttribute(CHANNEL_KEY); // 因为channle没有被关闭,这里能取到ret而不是重新创建channle

if (ret == null) {

ret = new HeaderExchangeChannel(ch);

if (ch.isConnected()) {

ch.setAttribute(CHANNEL_KEY, ret);

}

}

return ret;

}

`

出现这种问题的原因:

DefaultFuture 中有 CHANNELS ,在每次初始化 DefaultFuture 的时候会向里面添加数据

有用在 HeaderExchangeChannel中 close(int timeout) DefaultFuture.hasFuture(channel)

这里有一个循环,每隔10ms判断 CHANNELS 中是否还存在当前的 channel ,如果不存在则关闭当前channel,

这就导致可能会存在10ms的误差,期间 CHANNELS 中已经没有了当前channel,但是当前channel并未关闭,closed = true,

进而导致下一次调用该provider时,closed=true,在HeaderExchangeChannel 93行会抛出异常

原本的逻辑应该是HeaderExchangeChannel 的 63行,当rat不存在的时候 创建一个HeaderExchangeChannel 进行通信.

个人技术有限,现在没有完善的解决方案,希望各位大佬能解决一下

第一次写的乱、草率还请见谅