dubbo心跳机制 (2)
此文已由作者赵计刚授权网易云社区发布。
欢迎访问网易云社区,了解更多网易技术产品运营经验。
来看一下HeaderExchangeServer.this.getChannels():
1 public Collection<Channel> getChannels() {
2 return (Collection) getExchangeChannels();
3 }
4
5 public Collection<ExchangeChannel> getExchangeChannels() {
6 Collection<ExchangeChannel> exchangeChannels = new ArrayList<ExchangeChannel>();
7 Collection<Channel> channels = server.getChannels();
8 if (channels != null && channels.size() > 0) {
9 for (Channel channel : channels) {
10 exchangeChannels.add(HeaderExchangeChannel.getOrAddChannel(channel));
11 }
12 }
13 return exchangeChannels;
14 }
实际上就是获取NettyServer中的全部channel连接。
获取到需要心跳检测的channel后,对每一个channel进行如下判断:
如果在heartbeat内没有进行读操作或者写操作,则发送心跳请求
如果正常消息和心跳在heartbeatTimeout都没接收到,consumer端会进行重连,provider端会关闭channel
这里比较关键的是lastRead和lastWrite的设置。先来看一下获取:
1 Long lastRead = (Long) channel.getAttribute(HeaderExchangeHandler.KEY_READ_TIMESTAMP);
2 Long lastWrite = (Long) channel.getAttribute(HeaderExchangeHandler.KEY_WRITE_TIMESTAMP);
说明有地方在设置这两个值到channel中。
从请求和响应处理来看,无论是请求还是响应都会按照这个顺序处理一遍。
1 MultiMessageHandler
2 -->handler: HeartbeatHandler
3 -->handler: AllChannelHandler
4 -->url: providerUrl
5 -->executor: FixedExecutor
6 -->handler: DecodeHandler
7 -->handler: HeaderExchangeHandler
8 -->handler: ExchangeHandlerAdapter(DubboProtocol.requestHandler)
其中HeartbeatHandler源码如下:
1 public class HeartbeatHandler extends AbstractChannelHandlerDelegate {
2
3 private static final Logger logger = LoggerFactory.getLogger(HeartbeatHandler.class);
4
5 public static String KEY_READ_TIMESTAMP = "READ_TIMESTAMP";
6
7 public static String KEY_WRITE_TIMESTAMP = "WRITE_TIMESTAMP";
8
9 public HeartbeatHandler(ChannelHandler handler) {
10 super(handler);
11 }
12
13 public void connected(Channel channel) throws RemotingException {
14 setReadTimestamp(channel);
15 setWriteTimestamp(channel);
16 handler.connected(channel);
17 }
18
19 public void disconnected(Channel channel) throws RemotingException {
20 clearReadTimestamp(channel);
21 clearWriteTimestamp(channel);
22 handler.disconnected(channel);
23 }
24
25 public void sent(Channel channel, Object message) throws RemotingException {
26 setWriteTimestamp(channel);
27 handler.sent(channel, message);
28 }
29
30 public void received(Channel channel, Object message) throws RemotingException {
31 setReadTimestamp(channel);
32 if (isHeartbeatRequest(message)) {
33 Request req = (Request) message;
34 if (req.isTwoWay()) {
35 Response res = new Response(req.getId(), req.getVersion());
36 res.setEvent(Response.HEARTBEAT_EVENT);
37 channel.send(res);
38 if (logger.isInfoEnabled()) {
39 int heartbeat = channel.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
40 if (logger.isDebugEnabled()) {
41 logger.debug("Received heartbeat from remote channel " + channel.getRemoteAddress()
42 + ", cause: The channel has no data-transmission exceeds a heartbeat period"
43 + (heartbeat > 0 ? ": " + heartbeat + "ms" : ""));
44 }
45 }
46 }
47 return;
48 }
49 if (isHeartbeatResponse(message)) {
50 if (logger.isDebugEnabled()) {
51 logger.debug(
52 new StringBuilder(32)
53 .append("Receive heartbeat response in thread ")
54 .append(Thread.currentThread().getName())
55 .toString());
56 }
57 return;
58 }
59 handler.received(channel, message);
60 }
61
62 private void setReadTimestamp(Channel channel) {
63 channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
64 }
65
66 private void setWriteTimestamp(Channel channel) {
67 channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis());
68 }
69
70 private void clearReadTimestamp(Channel channel) {
71 channel.removeAttribute(KEY_READ_TIMESTAMP);
72 }
73
74 private void clearWriteTimestamp(Channel channel) {
75 channel.removeAttribute(KEY_WRITE_TIMESTAMP);
76 }
77
78 private boolean isHeartbeatRequest(Object message) {
79 return message instanceof Request && ((Request) message).isHeartbeat();
80 }
81
82 private boolean isHeartbeatResponse(Object message) {
83 return message instanceof Response && ((Response) message).isHeartbeat();
84 }
85 }
连接完成时:设置lastRead和lastWrite
连接断开时:清空lastRead和lastWrite
发送消息时:设置lastWrite
接收消息时:设置lastRead
之后交由AllChannelHandler进行处理。之后会一直交由HeaderExchangeHandler进行处理。其对lastRead和lastWrite也做了设置和清理:
1 public void connected(Channel channel) throws RemotingException {
2 channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
3 channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis()); 4 ...
5 }
6
7 public void disconnected(Channel channel) throws RemotingException {
8 channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
9 channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis());
10 ...
11 }
12
13 public void sent(Channel channel, Object message) throws RemotingException {
14 Throwable exception = null;
15 try {
16 channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis());
17 ...
18 } catch (Throwable t) {
19 exception = t;
20 }
21 }
22
23 public void received(Channel channel, Object message) throws RemotingException {
24 channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
25 ...
26 }
连接完成时:设置lastRead和lastWrite
连接断开时:也设置lastRead和lastWrite(为什么?)
发送消息时:设置lastWrite
接收消息时:设置lastRead
这里里有个疑问,从handler链来看,无论是请求还是响应都会按照handler链来处理一遍。那么在HeartbeatHandler中已经进行了lastWrite和lastRead的设置,为什么还要在HeaderExchangeHandler中再处理一遍?
最后,provider端认为连接断了,则会关闭channel。来看一下NettyChannel的close方法:
1 public void close() {
2 // 1 将close属性设为true
3 try {
4 super.close();
5 } catch (Exception e) {
6 logger.warn(e.getMessage(), e);
7 }
8 // 2 从全局NettyChannel缓存器中将当前的NettyChannel删掉
9 try {
10 removeChannelIfDisconnected(channel);
11 } catch (Exception e) {
12 logger.warn(e.getMessage(), e);
13 }
14 // 3 清空当前的NettyChannel中的attributes属性
15 try {
16 attributes.clear();
17 } catch (Exception e) {
18 logger.warn(e.getMessage(), e);
19 }
20 // 4 关闭netty的channel,执行netty的channel的优雅关闭
21 try {
22 if (logger.isInfoEnabled()) {
23 logger.info("Close netty channel " + channel);
24 }
25 channel.close();
26 } catch (Exception e) {
27 logger.warn(e.getMessage(), e);
28 }
29 }
从上边代码来看,假设consumer端挂了,provider端的心跳检测机制可以进行相关的资源回收,所以provider端的心跳检测机制是有必要的。
更多网易技术、产品、运营经验分享请点击。
相关文章:
【推荐】 大咖分享 | 一文解锁首届云创大会干货——下篇(文末附演讲ppt文件免费下载)
【推荐】 Kafka实践、升级和新版本(0.10)特性预研
最新文章
- thinkphp 3.2 join
- RealProxy实现AOP编程(2)
- 第三方Girdview中文件下载的方法,以及js显示图片
- Azure Blob
- dede模板完全控制攻略
- MyBatis学习总结_03_优化MyBatis配置文件中的配置
- QtWebkit2.2.0 HTML5.0支持情况
- Windows Server 2012的配置与部署
- PHP中利用redis实现消息队列处理高并发请求
- ajaxSetup设置Ajax请求的默认值
- Filecoin:募资详情和Token分发详情
- java @Override 报错解决
- git 拉取远程代码
- 每天一条linux命令
- c# Resolve SQlite Concurrency Exception Problem (Using Read-Write Lock)
- mdadm命令详解
- Cookies与session的区别
- tkinter 对键盘和鼠标事件的处理
- Jquery DataTable基本使用
- js中文乱码问题,编码设为utf-8,但还是乱码问题。