- Java锁的逻辑(结合对象头和ObjectMonitor)
- 还在用饼状图?来瞧瞧这些炫酷的百分比可视化新图形(附代码实现)⛵
- 自动注册实体类到EntityFrameworkCore上下文,并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
Dubbo源码阅读分享系列文章,欢迎大家关注点赞 。 在网络交互中是以字节流的形式传递的,对于字节流都是二进制格式,这样我们就面临一个问题就是如何转化为我们可以识别的字符,协议就是来解决这个问题的,协议用通俗易懂地解释就是通信双方需要遵循的约定。 在日常开发中,我们常见的网络传输协议有TCP、UDP、HTTP。常用的中间件也会定义对应的协议,如Redis、Mysql、Zookeeper等都有自己约定的协议,同样Dubbo的通信也采用一种协议,这些都是应用层协议,都是基于TCP或者UDP设计的. 应用层协议一般的形式有三种:定长协议、特殊结束符和变长协议,聊到这里就可以抛出来一个常见的面试题,如何解决网络通信粘包和拆包的问题?该问题的解决方案也就是通过约定协议,下面我们就来聊聊这三种模式优缺点以及使用场景. 定长的协议是指协议内容的长度是固定的,比如协议byte长度是50,当从网络上读取50个byte后,就进行decode解码操作. 定长协议在读取或者写入时,效率比较高,因为数据大小都是确定的. 定长协议的缺点在于适应性不足,网络传输中传输的内容的大小不可能都是相同的,因此对于一些长度不够的消息,明显过于的浪费带宽. 特殊结束符就是在每次传输结束的时候使用一个特殊的结束符,在Redis中的协议采用了特殊结束符,客户端和服务器发送的命令一律使用\r\n(CRLF)结尾. 与定长协议一样读取或者写入时,效率比较高,同时解决定长协议的尴尬. 特殊结束符方式的问题是必须要有一个完整的消息体才能进行传输,除此之外必须要防止用户传输的数据不能同结束符相同,否则就会出现紊乱. 变长协议由定长以及不定长两部分组成,定长部分一般是协议头,此部分会包含变长部分的描述,变长协议我们经常使用的HTTP协议采用变长协议,HTTP请求报文格式是由三部分组成: 灵活性比较高,解决了定长协议以及特殊结束符的所有缺点. 复杂性比较高,需要自定义一套标准,所有消息都需要按照该格式发送以及解析. Dubbo框架支持很多协议,默认采用Dubbo协议,Dubbo协议采用的是变长协议的设计,整体的格式如下 Dubbo协议整体设计比较简洁,能采用1个bit表示的,不会用一个byte来表示;此外请求头和响应头一致,整体采用一套解析标准就可以,代码实现起来相对简单. 由于整体的设计相对简洁,导致扩展性不够; 。 在通信篇中我们讲过Codec2该接口,该接口提供了encode和decode个方法来实现消息与字节流之间的相互转换,关于该接口的实现我们没有讲解,这里我们来看看此部分和Dubbo协议有什么关系。 AbstractCodec抽象类没有实现Codec2中定义的接口方法,而是提供了几个给子类用的基础方法. 接下来我们就来聊聊子类如何被解析的,我们可以看到四个子类的继承关系,重点介绍的是ExchangeCodec和DubboCodec,其他就是做一下简单介绍。 TransportCodec该类已经被标注为弃用,该类内部也就是根据getSerialization方法选择的序列化方法,对传入消息或ChannelBuffer进行序列化或反序列化。 TelnetCodec继承了TransportCodec的能力,该类主要是提供了对Telnet命令处理的能力,该功能主要是对服务进行治理的功能,这里后续我们画一点时间来进行介绍. ExchangeCodec继承了TelnetCodec,在该类基础上增加Dubbo协议头的处理能力,接下来我们首先来看下其核心字段, 。 通过核心字段我们可以发现其实和我们介绍的Dubbo的协议是一致的,因此接下来的encode和decode就是对Dubbo协议头的解密和编码,我们来下看encode方法,在encode方法中会根据需要编码的消息类型进行分类, 分为三类:Request、Response、telenet,encodeRequest方法专门对Request对象进行编码,encodeResponse方法对Response对象进行编码. ExchangeCodec的decode方法是encode方法的逆过程,会先检查魔数,然后读取协议头和后续消息的长度,最后根据协议头中的各个标志位构造相应的对象,以及反序列化数据. 在ExchangeCodecencode的encode方法中,不论是encodeRequest还是encodeResponse都调用encodeRequestData方法,该方法会对Boby内容进行编码,该方法实现是在DubboCodec中,因此DubboCodec是对消息体的编解码,接下来我们来看下encodeRequestData和encodeResponseData方法的实现, 。 欢迎大家点点关注,点点赞. 前言
SPI实现部分
注册中心
通信
什么是协议
如何定义协议
定长协议
优点
缺点
特殊结束符
优点
缺点
变长协议
优点
缺点
Dubbo协议
优点
缺点
Dubbo协议是如何解析的
ExchangeCodec
//协议头长度
protected
static
final
int
HEADER_LENGTH =
16
;
//魔数 判断是否是Dubbo协议
protected
static
final
short
MAGIC = (
short
)
0xdabb
;
protected
static
final
byte
MAGIC_HIGH = Bytes.short2bytes(MAGIC)[
0
];
protected
static
final
byte
MAGIC_LOW = Bytes.short2bytes(MAGIC)[
1
];
//设置请求响应标志位
protected
static
final
byte
FLAG_REQUEST = (
byte
)
0x80
;
//单向还是双向标志位
protected
static
final
byte
FLAG_TWOWAY = (
byte
)
0x40
;
//是否事件消息标志位
protected
static
final
byte
FLAG_EVENT = (
byte
)
0x20
;
//序列化协议标志位
protected
static
final
int
SERIALIZATION_MASK =
0x1f
;
@Override
public void encode (Channel channel, ChannelBuffer buffer, Object msg) throws IOException
{
//Request
if
(msg
instanceof
Request) {
encodeRequest(channel, buffer, (Request) msg);
//Response
}
else
if
(msg
instanceof
Response) {
encodeResponse(channel, buffer, (Response) msg);
}
else
{
//telenet
super
.encode(channel, buffer, msg);
}
}
protected void encodeRequest (Channel channel, ChannelBuffer buffer, Request req) throws IOException
{
Serialization serialization = getSerialization(channel, req);
//存储协议头
byte
[] header =
new
byte
[HEADER_LENGTH];
// set magic number.
Bytes.short2bytes(MAGIC, header);
//设置协议头标志位
header[
2
] = (
byte
) (FLAG_REQUEST | serialization.getContentTypeId());
if
(req.isTwoWay()) {
header[
2
] |= FLAG_TWOWAY;
}
if
(req.isEvent()) {
header[
2
] |= FLAG_EVENT;
}
//记录请求ID
Bytes.long2bytes(req.getId(), header,
4
);
//序列化请求 并统计序列化以后字节数
int
savedWriteIndex = buffer.writerIndex();
//将写入位置后移16位
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
//请求序列化
ChannelBufferOutputStream bos =
new
ChannelBufferOutputStream(buffer);
//是否心跳检查 为空就是心跳检查
if
(req.isHeartbeat()) {
// heartbeat request data is always null
bos.write(CodecSupport.getNullBytesOf(serialization));
}
else
{
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
//事件序列化
if
(req.isEvent()) {
//事件序列化
encodeEventData(channel, out, req.getData());
}
else
{
//Dubbo请求序列化
encodeRequestData(channel, out, req.getData(), req.getVersion());
}
out.flushBuffer();
if
(out
instanceof
Cleanable) {
((Cleanable) out).cleanup();
}
}
bos.flush();
bos.close();
//获取字节数
int
len = bos.writtenBytes();
//检查字节长度
checkPayload(channel, len);
//将字节数写入header数组中
Bytes.int2bytes(len, header,
12
);
//重置写入位置
buffer.writerIndex(savedWriteIndex);
//写入消息头
buffer.writeBytes(header);
//buffer写出去的位置从writeIndex开始 加上header长度 数据长度
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}
protected void encodeResponse (Channel channel, ChannelBuffer buffer, Response res) throws IOException
{
int
savedWriteIndex = buffer.writerIndex();
try
{
//序列化
Serialization serialization = getSerialization(channel, res);
//协议头 长度为16字节
byte
[] header =
new
byte
[HEADER_LENGTH];
//魔数
Bytes.short2bytes(MAGIC, header);
//序列化方式
header[
2
] = serialization.getContentTypeId();
//心跳还是正常消息
if
(res.isHeartbeat()) {
header[
2
] |= FLAG_EVENT;
}
//响应状态
byte
status = res.getStatus();
header[
3
] = status;
//设置请求ID
Bytes.long2bytes(res.getId(), header,
4
);
//写入时候真需要加上协议头长度
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
ChannelBufferOutputStream bos =
new
ChannelBufferOutputStream(buffer);
//对响应信息进行编码
if
(status == Response.OK) {
if
(res.isHeartbeat()){
//心跳
bos.write(CodecSupport.getNullBytesOf(serialization));
}
else
{
//正常响应
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
if
(res.isEvent()) {
encodeEventData(channel, out, res.getResult());
}
else
{
encodeResponseData(channel, out, res.getResult(), res.getVersion());
}
out.flushBuffer();
if
(out
instanceof
Cleanable) {
((Cleanable) out).cleanup();
}
}
}
else
{
//错误消息
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
out.writeUTF(res.getErrorMessage());
out.flushBuffer();
if
(out
instanceof
Cleanable) {
((Cleanable) out).cleanup();
}
}
bos.flush();
bos.close();
//写入的长度
int
len = bos.writtenBytes();
//检查消息长度
checkPayload(channel, len);
Bytes.int2bytes(len, header,
12
);
//重置写入位置
buffer.writerIndex(savedWriteIndex);
//写入消息头
buffer.writeBytes(header);
//buffer写出去的位置从writeIndex开始 加上header长度 数据长度
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}
catch
(Throwable t) {
// clear buffer
buffer.writerIndex(savedWriteIndex);
// send error message to Consumer, otherwise, Consumer will wait till timeout.
if
(!res.isEvent() && res.getStatus() != Response.BAD_RESPONSE) {
Response r =
new
Response(res.getId(), res.getVersion());
r.setStatus(Response.BAD_RESPONSE);
if
(t
instanceof
ExceedPayloadLimitException) {
logger.warn(t.getMessage(), t);
try
{
r.setErrorMessage(t.getMessage());
channel.send(r);
return
;
}
catch
(RemotingException e) {
logger.warn(
"Failed to send bad_response info back: "
+ t.getMessage() +
", cause: "
+ e.getMessage(), e);
}
}
else
{
// FIXME log error message in Codec and handle in caught() of IoHanndler?
logger.warn(
"Fail to encode response: "
+ res +
", send bad_response info instead, cause: "
+ t.getMessage(), t);
try
{
r.setErrorMessage(
"Failed to send response: "
+ res +
", cause: "
+ StringUtils.toString(t));
channel.send(r);
return
;
}
catch
(RemotingException e) {
logger.warn(
"Failed to send bad_response info back: "
+ res +
", cause: "
+ e.getMessage(), e);
}
}
}
// Rethrow exception
if
(t
instanceof
IOException) {
throw
(IOException) t;
}
else
if
(t
instanceof
RuntimeException) {
throw
(RuntimeException) t;
}
else
if
(t
instanceof
Error) {
throw
(Error) t;
}
else
{
throw
new
RuntimeException(t.getMessage(), t);
}
}
}
DubboCodec
protected void encodeRequestData (Channel channel, ObjectOutput out, Object data, String version) throws IOException
{
RpcInvocation inv = (RpcInvocation) data;
//dubbo服务版本
out.writeUTF(version);
// https://github.com/apache/dubbo/issues/6138
String serviceName = inv.getAttachment(INTERFACE_KEY);
if
(serviceName ==
null
) {
//服务path
serviceName = inv.getAttachment(PATH_KEY);
}
//服务名
out.writeUTF(serviceName);
//版本号
out.writeUTF(inv.getAttachment(VERSION_KEY));
//方法名
out.writeUTF(inv.getMethodName());
//方法类型描述
out.writeUTF(inv.getParameterTypesDesc());
Object[] args = inv.getArguments();
if
(args !=
null
) {
for
(
int
i =
0
; i < args.length; i++) {
//参数值
out.writeObject(encodeInvocationArgument(channel, inv, i));
}
}
//附加属性
out.writeAttachments(inv.getObjectAttachments());
}
@Override
protected void encodeResponseData (Channel channel, ObjectOutput out, Object data, String version) throws IOException
{
Result result = (Result) data;
//检验版本
boolean
attach = Version.isSupportResponseAttachment(version);
Throwable th = result.getException();
if
(th ==
null
) {
Object ret = result.getValue();
if
(ret ==
null
) {
//空结果
out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE);
}
else
{
//正常写入
out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE);
out.writeObject(ret);
}
}
else
{
//异常
out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION);
out.writeThrowable(th);
}
if
(attach) {
//Dubbo版本号
result.getObjectAttachments().put(DUBBO_VERSION_KEY, Version.getProtocolVersion());
out.writeAttachments(result.getObjectAttachments());
}
}
结束
最后此篇关于Dubbo-聊聊Dubbo协议的文章就讲到这里了,如果你想了解更多关于Dubbo-聊聊Dubbo协议的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
有多久,没有发过短信了? 1、背景简介 在常规的分布式架构下,「消息中心」的服务里通常会集成「短信」的渠道,作为信息触达的重要手段,其他常用的手段还包括:「某微」、「某钉」、
区块链、低代码、元宇宙、AI智能; 01 【 先来说说背景 】 这个概念由来已久,但是在国内兴起,是最近几年; 低代码即「 Low-Code
目录 1、背景简介 2、订单业务 1、订单体系 2、流程管理 2
1. 相同点 用Python语言编写的源代码文件,其文件后缀是 “.py” 或 “.ipynb”。用Python语言编写的源代码文件,其文件后缀是 “.py” 或 “.ipynb”。 2. 区别
功能简介 闭锁是一种同步工具类,可以延迟线程的进度直到其到达终止状态【CPJ 3.4.2】。闭锁的作用相当于一扇门∶ 在闭锁到达结束状态之前,这扇门一直是关闭的,并且没有任何线程能通过,当到达
高阶函数,英文叫 Higher Order function。一个函数可以接收另外一个函数作为参数,这种函数就叫做高阶函数。 示例: function add(x, 
引文 最近公司项目中使用了 Nuxt 框架,进行首屏的服务端渲染,加快了内容的到达时间 (time-to-content),于是笔者开始了对 Nuxt 的学习和使用。以下是从源码角度对 Nux
什么是游标? 游标(cursor)是一个存储在MySQL服务器上的数据库查询, 它不是一条SELECT语句,而是被该语句检索出来的结果集。在存储了游 标之后,应用程序可以根据需要滚动或浏
流水线工作模型在工业领域内十分常见,它将工作流程分为多个环节,每个环节根据工作强度安排合适的人员数量。良好的流水线设计尽量让各环节的流通率平衡,最大化提高产能效率。 Go 是一门实用性语言,流
1. Spring JDBC Spring JDBC的配置 2. Spring JdbcTemplate的常用方法 execute()
1. 前言 大家好,我是安果! 日常编写的 Python 自动化程序,如果在本地运行稳定后,就可以考虑将它部署到服务器,结合定时任务完全解放双手 但是,由于自动化程序与平台兼
前言 有时候我们会有在需要在网页中写代码或者改代码配置的需求,这个时候就需要用到代码编辑器,常规的代码编辑器有 CodeMirror 和 Monaco Editor, CodeMirror 使用的
前言:模块机制是 Node.js 中非常重要的组成,模块机制使得我们可以以模块化的方式写代码,而不是全部代码都写到一个文件里。我们平时使用的比较多的通过 require 加载模块,但是我们可能不
随着互联网的发展,越来越多的公司摒弃了Hibernate,而选择拥抱了MyBatis。而且,很多大厂在面试的时候喜欢问MyBatis底层的原理和源码实现。 总之,MyBatis几乎成为了Jav
@requestmapping和@getmapping @postmapping的区别 最近学习看一些代码,发现对于发送请求这件事,有的地方用@requestmapping,有的地方用@postm
@RequestParam 和 @PathVariable 注解是用于从request中接收请求的,两个都可以接收参数,关键点不同的是@RequestParam 是从request里面拿取值,而 @
PHP8的Alpha版本,过几天就要发布了,其中包含了不少的新特性,当然我自己认为最重要的还是JIT,这个我从2013年开始参与,中间挫折无数,失败无数后,终于要发布的东东。 不过,今天呢,我不打
引言 我们想要网格的服务发现、路由、熔断降级、负载均衡,这些流量治理都在数据面Envoy中执行才行。Envoy也提供的Filter机制来做这些功能,通常有以下方式: 通过C
我是一名优秀的程序员,十分优秀!