Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
西红柿炒蛋不加盐
guide-rpc-framework
提交
5959b8ee
G
guide-rpc-framework
项目概览
西红柿炒蛋不加盐
/
guide-rpc-framework
与 Fork 源项目一致
Fork自
嗝屁小孩纸 / guide-rpc-framework
通知
1
Star
1
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
G
guide-rpc-framework
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
提交
5959b8ee
编写于
11月 19, 2020
作者:
G
guide
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[refractor]主要重构了网络传输部分的代码
上级
801287f4
变更
21
隐藏空白更改
内联
并排
Showing
21 changed file
with
116 addition
and
137 deletion
+116
-137
README-EN.md
README-EN.md
+2
-2
README.md
README.md
+2
-2
docs/使用CompletableFuture优化接受服务提供端返回结果.md
docs/使用CompletableFuture优化接受服务提供端返回结果.md
+1
-1
example-client/src/main/java/github/javaguide/SocketClientMain.java
...ient/src/main/java/github/javaguide/SocketClientMain.java
+4
-4
example-server/src/main/java/NettyServerMain.java
example-server/src/main/java/NettyServerMain.java
+4
-4
example-server/src/main/java/SocketServerMain.java
example-server/src/main/java/SocketServerMain.java
+1
-1
rpc-framework-simple/src/main/java/github/javaguide/provider/ServiceProviderImpl.java
...n/java/github/javaguide/provider/ServiceProviderImpl.java
+2
-2
rpc-framework-simple/src/main/java/github/javaguide/proxy/RpcClientProxy.java
.../src/main/java/github/javaguide/proxy/RpcClientProxy.java
+11
-11
rpc-framework-simple/src/main/java/github/javaguide/registry/zk/util/CuratorUtils.java
.../java/github/javaguide/registry/zk/util/CuratorUtils.java
+3
-5
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/RpcRequestTransport.java
...hub/javaguide/remoting/transport/RpcRequestTransport.java
+1
-1
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/ChannelProvider.java
...uide/remoting/transport/netty/client/ChannelProvider.java
+5
-6
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyClientTransport.java
...remoting/transport/netty/client/NettyClientTransport.java
+0
-74
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyRpcClient.java
...guide/remoting/transport/netty/client/NettyRpcClient.java
+63
-6
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyRpcClientHandler.java
...emoting/transport/netty/client/NettyRpcClientHandler.java
+5
-5
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/server/NettyRpcServer.java
...guide/remoting/transport/netty/server/NettyRpcServer.java
+2
-2
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/server/NettyRpcServerHandler.java
...emoting/transport/netty/server/NettyRpcServerHandler.java
+2
-2
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/socket/SocketRpcClient.java
.../javaguide/remoting/transport/socket/SocketRpcClient.java
+2
-2
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/socket/SocketRpcServer.java
.../javaguide/remoting/transport/socket/SocketRpcServer.java
+1
-2
rpc-framework-simple/src/main/java/github/javaguide/spring/SpringBeanPostProcessor.java
...java/github/javaguide/spring/SpringBeanPostProcessor.java
+3
-3
rpc-framework-simple/src/main/resources/META-INF/extensions/github.javaguide.remoting.transport.ClientTransport
...sions/github.javaguide.remoting.transport.ClientTransport
+0
-2
rpc-framework-simple/src/main/resources/META-INF/extensions/github.javaguide.remoting.transport.RpcRequestTransport
...s/github.javaguide.remoting.transport.RpcRequestTransport
+2
-0
未找到文件。
README-EN.md
浏览文件 @
5959b8ee
...
...
@@ -181,10 +181,10 @@ public class HelloController {
```
```
java
ClientTransport
clien
tTransport
=
new
SocketRpcClient
();
ClientTransport
rpcReques
tTransport
=
new
SocketRpcClient
();
RpcServiceProperties
rpcServiceProperties
=
RpcServiceProperties
.
builder
()
.
group
(
"test2"
).
version
(
"version2"
).
build
();
RpcClientProxy
rpcClientProxy
=
new
RpcClientProxy
(
clien
tTransport
,
rpcServiceProperties
);
RpcClientProxy
rpcClientProxy
=
new
RpcClientProxy
(
rpcReques
tTransport
,
rpcServiceProperties
);
HelloService
helloService
=
rpcClientProxy
.
getProxy
(
HelloService
.
class
);
String
hello
=
helloService
.
hello
(
new
Hello
(
"111"
,
"222"
));
System
.
out
.
println
(
hello
);
...
...
README.md
浏览文件 @
5959b8ee
...
...
@@ -226,10 +226,10 @@ public class HelloController {
```
```
java
ClientTransport
clien
tTransport
=
new
SocketRpcClient
();
ClientTransport
rpcReques
tTransport
=
new
SocketRpcClient
();
RpcServiceProperties
rpcServiceProperties
=
RpcServiceProperties
.
builder
()
.
group
(
"test2"
).
version
(
"version2"
).
build
();
RpcClientProxy
rpcClientProxy
=
new
RpcClientProxy
(
clien
tTransport
,
rpcServiceProperties
);
RpcClientProxy
rpcClientProxy
=
new
RpcClientProxy
(
rpcReques
tTransport
,
rpcServiceProperties
);
HelloService
helloService
=
rpcClientProxy
.
getProxy
(
HelloService
.
class
);
String
hello
=
helloService
.
hello
(
new
Hello
(
"111"
,
"222"
));
System
.
out
.
println
(
hello
);
...
...
docs/使用CompletableFuture优化接受服务提供端返回结果.md
浏览文件 @
5959b8ee
...
...
@@ -33,7 +33,7 @@
现在,在你只需要通过下面的方式就能成功接收到客户端返回的结果:
```
java
CompletableFuture
<
RpcResponse
>
completableFuture
=
(
CompletableFuture
<
RpcResponse
>)
clien
tTransport
.
sendRpcRequest
(
rpcRequest
);
CompletableFuture
<
RpcResponse
>
completableFuture
=
(
CompletableFuture
<
RpcResponse
>)
rpcReques
tTransport
.
sendRpcRequest
(
rpcRequest
);
rpcResponse
=
completableFuture
.
get
();
```
example-client/src/main/java/github/javaguide/
RpcFrameworkSimple
ClientMain.java
→
example-client/src/main/java/github/javaguide/
Socket
ClientMain.java
浏览文件 @
5959b8ee
...
...
@@ -2,19 +2,19 @@ package github.javaguide;
import
github.javaguide.entity.RpcServiceProperties
;
import
github.javaguide.proxy.RpcClientProxy
;
import
github.javaguide.remoting.transport.
Clien
tTransport
;
import
github.javaguide.remoting.transport.
RpcReques
tTransport
;
import
github.javaguide.remoting.transport.socket.SocketRpcClient
;
/**
* @author shuang.kou
* @createTime 2020年05月10日 07:25:00
*/
public
class
RpcFrameworkSimple
ClientMain
{
public
class
Socket
ClientMain
{
public
static
void
main
(
String
[]
args
)
{
ClientTransport
clien
tTransport
=
new
SocketRpcClient
();
RpcRequestTransport
rpcReques
tTransport
=
new
SocketRpcClient
();
RpcServiceProperties
rpcServiceProperties
=
RpcServiceProperties
.
builder
()
.
group
(
"test2"
).
version
(
"version2"
).
build
();
RpcClientProxy
rpcClientProxy
=
new
RpcClientProxy
(
clien
tTransport
,
rpcServiceProperties
);
RpcClientProxy
rpcClientProxy
=
new
RpcClientProxy
(
rpcReques
tTransport
,
rpcServiceProperties
);
HelloService
helloService
=
rpcClientProxy
.
getProxy
(
HelloService
.
class
);
String
hello
=
helloService
.
hello
(
new
Hello
(
"111"
,
"222"
));
System
.
out
.
println
(
hello
);
...
...
example-server/src/main/java/NettyServerMain.java
浏览文件 @
5959b8ee
import
github.javaguide.HelloService
;
import
github.javaguide.annotation.RpcScan
;
import
github.javaguide.entity.RpcServiceProperties
;
import
github.javaguide.remoting.transport.netty.server.NettyServer
;
import
github.javaguide.remoting.transport.netty.server.Netty
Rpc
Server
;
import
github.javaguide.serviceimpl.HelloServiceImpl2
;
import
org.springframework.context.annotation.AnnotationConfigApplicationContext
;
...
...
@@ -16,12 +16,12 @@ public class NettyServerMain {
public
static
void
main
(
String
[]
args
)
{
// Register service via annotation
AnnotationConfigApplicationContext
applicationContext
=
new
AnnotationConfigApplicationContext
(
NettyServerMain
.
class
);
Netty
Server
nettyServer
=
(
NettyServer
)
applicationContext
.
getBean
(
"netty
Server"
);
Netty
RpcServer
nettyRpcServer
=
(
NettyRpcServer
)
applicationContext
.
getBean
(
"nettyRpc
Server"
);
// Register service manually
HelloService
helloService2
=
new
HelloServiceImpl2
();
RpcServiceProperties
rpcServiceProperties
=
RpcServiceProperties
.
builder
()
.
group
(
"test2"
).
version
(
"version2"
).
build
();
nettyServer
.
registerService
(
helloService2
,
rpcServiceProperties
);
nettyServer
.
start
();
netty
Rpc
Server
.
registerService
(
helloService2
,
rpcServiceProperties
);
netty
Rpc
Server
.
start
();
}
}
example-server/src/main/java/
RpcFrameworkSimple
ServerMain.java
→
example-server/src/main/java/
Socket
ServerMain.java
浏览文件 @
5959b8ee
...
...
@@ -7,7 +7,7 @@ import github.javaguide.serviceimpl.HelloServiceImpl;
* @author shuang.kou
* @createTime 2020年05月10日 07:25:00
*/
public
class
RpcFrameworkSimple
ServerMain
{
public
class
Socket
ServerMain
{
public
static
void
main
(
String
[]
args
)
{
HelloService
helloService
=
new
HelloServiceImpl
();
SocketRpcServer
socketRpcServer
=
new
SocketRpcServer
();
...
...
rpc-framework-simple/src/main/java/github/javaguide/provider/ServiceProviderImpl.java
浏览文件 @
5959b8ee
...
...
@@ -5,7 +5,7 @@ import github.javaguide.enums.RpcErrorMessageEnum;
import
github.javaguide.exception.RpcException
;
import
github.javaguide.extension.ExtensionLoader
;
import
github.javaguide.registry.ServiceRegistry
;
import
github.javaguide.remoting.transport.netty.server.NettyServer
;
import
github.javaguide.remoting.transport.netty.server.Netty
Rpc
Server
;
import
lombok.extern.slf4j.Slf4j
;
import
java.net.InetAddress
;
...
...
@@ -70,7 +70,7 @@ public class ServiceProviderImpl implements ServiceProvider {
String
serviceName
=
serviceRelatedInterface
.
getCanonicalName
();
rpcServiceProperties
.
setServiceName
(
serviceName
);
this
.
addService
(
service
,
serviceRelatedInterface
,
rpcServiceProperties
);
serviceRegistry
.
registerService
(
rpcServiceProperties
.
toRpcServiceName
(),
new
InetSocketAddress
(
host
,
NettyServer
.
PORT
));
serviceRegistry
.
registerService
(
rpcServiceProperties
.
toRpcServiceName
(),
new
InetSocketAddress
(
host
,
Netty
Rpc
Server
.
PORT
));
}
catch
(
UnknownHostException
e
)
{
log
.
error
(
"occur exception when getHostAddress"
,
e
);
}
...
...
rpc-framework-simple/src/main/java/github/javaguide/proxy/RpcClientProxy.java
浏览文件 @
5959b8ee
...
...
@@ -6,8 +6,8 @@ import github.javaguide.enums.RpcResponseCodeEnum;
import
github.javaguide.exception.RpcException
;
import
github.javaguide.remoting.dto.RpcRequest
;
import
github.javaguide.remoting.dto.RpcResponse
;
import
github.javaguide.remoting.transport.
Clien
tTransport
;
import
github.javaguide.remoting.transport.netty.client.Netty
ClientTranspor
t
;
import
github.javaguide.remoting.transport.
RpcReques
tTransport
;
import
github.javaguide.remoting.transport.netty.client.Netty
RpcClien
t
;
import
github.javaguide.remoting.transport.socket.SocketRpcClient
;
import
lombok.SneakyThrows
;
import
lombok.extern.slf4j.Slf4j
;
...
...
@@ -34,11 +34,11 @@ public class RpcClientProxy implements InvocationHandler {
/**
* Used to send requests to the server.And there are two implementations: socket and netty
*/
private
final
ClientTransport
clien
tTransport
;
private
final
RpcRequestTransport
rpcReques
tTransport
;
private
final
RpcServiceProperties
rpcServiceProperties
;
public
RpcClientProxy
(
ClientTransport
clien
tTransport
,
RpcServiceProperties
rpcServiceProperties
)
{
this
.
clientTransport
=
clien
tTransport
;
public
RpcClientProxy
(
RpcRequestTransport
rpcReques
tTransport
,
RpcServiceProperties
rpcServiceProperties
)
{
this
.
rpcRequestTransport
=
rpcReques
tTransport
;
if
(
rpcServiceProperties
.
getGroup
()
==
null
)
{
rpcServiceProperties
.
setGroup
(
""
);
}
...
...
@@ -49,8 +49,8 @@ public class RpcClientProxy implements InvocationHandler {
}
public
RpcClientProxy
(
ClientTransport
clien
tTransport
)
{
this
.
clientTransport
=
clien
tTransport
;
public
RpcClientProxy
(
RpcRequestTransport
rpcReques
tTransport
)
{
this
.
rpcRequestTransport
=
rpcReques
tTransport
;
this
.
rpcServiceProperties
=
RpcServiceProperties
.
builder
().
group
(
""
).
version
(
""
).
build
();
}
...
...
@@ -80,12 +80,12 @@ public class RpcClientProxy implements InvocationHandler {
.
version
(
rpcServiceProperties
.
getVersion
())
.
build
();
RpcResponse
<
Object
>
rpcResponse
=
null
;
if
(
clientTransport
instanceof
NettyClientTranspor
t
)
{
CompletableFuture
<
RpcResponse
<
Object
>>
completableFuture
=
(
CompletableFuture
<
RpcResponse
<
Object
>>)
clien
tTransport
.
sendRpcRequest
(
rpcRequest
);
if
(
rpcRequestTransport
instanceof
NettyRpcClien
t
)
{
CompletableFuture
<
RpcResponse
<
Object
>>
completableFuture
=
(
CompletableFuture
<
RpcResponse
<
Object
>>)
rpcReques
tTransport
.
sendRpcRequest
(
rpcRequest
);
rpcResponse
=
completableFuture
.
get
();
}
if
(
clien
tTransport
instanceof
SocketRpcClient
)
{
rpcResponse
=
(
RpcResponse
<
Object
>)
clien
tTransport
.
sendRpcRequest
(
rpcRequest
);
if
(
rpcReques
tTransport
instanceof
SocketRpcClient
)
{
rpcResponse
=
(
RpcResponse
<
Object
>)
rpcReques
tTransport
.
sendRpcRequest
(
rpcRequest
);
}
this
.
check
(
rpcResponse
,
rpcRequest
);
return
rpcResponse
.
getData
();
...
...
rpc-framework-simple/src/main/java/github/javaguide/registry/zk/util/CuratorUtils.java
浏览文件 @
5959b8ee
...
...
@@ -33,7 +33,7 @@ public final class CuratorUtils {
private
static
final
Map
<
String
,
List
<
String
>>
SERVICE_ADDRESS_MAP
=
new
ConcurrentHashMap
<>();
private
static
final
Set
<
String
>
REGISTERED_PATH_SET
=
ConcurrentHashMap
.
newKeySet
();
private
static
CuratorFramework
zkClient
;
private
static
String
defaultZookeeperAddress
=
"127.0.0.1:2181"
;
private
static
final
String
DEFAULT_ZOOKEEPER_ADDRESS
=
"127.0.0.1:2181"
;
private
CuratorUtils
()
{
}
...
...
@@ -97,9 +97,7 @@ public final class CuratorUtils {
public
static
CuratorFramework
getZkClient
()
{
// check if user has set zk address
Properties
properties
=
PropertiesFileUtil
.
readPropertiesFile
(
RpcConfigEnum
.
RPC_CONFIG_PATH
.
getPropertyValue
());
if
(
properties
!=
null
)
{
defaultZookeeperAddress
=
properties
.
getProperty
(
RpcConfigEnum
.
ZK_ADDRESS
.
getPropertyValue
());
}
String
zookeeperAddress
=
properties
!=
null
&&
properties
.
getProperty
(
RpcConfigEnum
.
ZK_ADDRESS
.
getPropertyValue
())
!=
null
?
properties
.
getProperty
(
RpcConfigEnum
.
ZK_ADDRESS
.
getPropertyValue
())
:
DEFAULT_ZOOKEEPER_ADDRESS
;
// if zkClient has been started, return directly
if
(
zkClient
!=
null
&&
zkClient
.
getState
()
==
CuratorFrameworkState
.
STARTED
)
{
return
zkClient
;
...
...
@@ -108,7 +106,7 @@ public final class CuratorUtils {
RetryPolicy
retryPolicy
=
new
ExponentialBackoffRetry
(
BASE_SLEEP_TIME
,
MAX_RETRIES
);
zkClient
=
CuratorFrameworkFactory
.
builder
()
// the server to connect to (can be a server list)
.
connectString
(
defaultZ
ookeeperAddress
)
.
connectString
(
z
ookeeperAddress
)
.
retryPolicy
(
retryPolicy
)
.
build
();
zkClient
.
start
();
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/
Clien
tTransport.java
→
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/
RpcReques
tTransport.java
浏览文件 @
5959b8ee
...
...
@@ -10,7 +10,7 @@ import github.javaguide.remoting.dto.RpcRequest;
* @createTime 2020年05月29日 13:26:00
*/
@SPI
public
interface
Clien
tTransport
{
public
interface
RpcReques
tTransport
{
/**
* send rpc request to server and get result
*
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/ChannelProvider.java
浏览文件 @
5959b8ee
package
github.javaguide.remoting.transport.netty.client
;
import
github.javaguide.factory.SingletonFactory
;
import
io.netty.channel.Channel
;
import
lombok.extern.slf4j.Slf4j
;
...
...
@@ -18,11 +17,9 @@ import java.util.concurrent.ConcurrentHashMap;
public
class
ChannelProvider
{
private
final
Map
<
String
,
Channel
>
channelMap
;
private
final
NettyClient
nettyClient
;
public
ChannelProvider
()
{
channelMap
=
new
ConcurrentHashMap
<>();
nettyClient
=
SingletonFactory
.
getInstance
(
NettyClient
.
class
);
}
public
Channel
get
(
InetSocketAddress
inetSocketAddress
)
{
...
...
@@ -37,10 +34,12 @@ public class ChannelProvider {
channelMap
.
remove
(
key
);
}
}
// otherwise, reconnect to get the Channel
Channel
channel
=
nettyClient
.
doConnect
(
inetSocketAddress
);
return
null
;
}
public
void
set
(
InetSocketAddress
inetSocketAddress
,
Channel
channel
)
{
String
key
=
inetSocketAddress
.
toString
();
channelMap
.
put
(
key
,
channel
);
return
channel
;
}
public
void
remove
(
InetSocketAddress
inetSocketAddress
)
{
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyClientTransport.java
已删除
100644 → 0
浏览文件 @
801287f4
package
github.javaguide.remoting.transport.netty.client
;
import
github.javaguide.enums.CompressTypeEnum
;
import
github.javaguide.extension.ExtensionLoader
;
import
github.javaguide.factory.SingletonFactory
;
import
github.javaguide.registry.ServiceDiscovery
;
import
github.javaguide.remoting.constants.RpcConstants
;
import
github.javaguide.remoting.dto.RpcMessage
;
import
github.javaguide.remoting.dto.RpcRequest
;
import
github.javaguide.remoting.dto.RpcResponse
;
import
github.javaguide.remoting.transport.ClientTransport
;
import
github.javaguide.enums.SerializationTypeEnum
;
import
io.netty.channel.Channel
;
import
io.netty.channel.ChannelFutureListener
;
import
lombok.extern.slf4j.Slf4j
;
import
java.net.InetSocketAddress
;
import
java.util.concurrent.CompletableFuture
;
/**
* transport rpcRequest based on netty.
*
* @author shuang.kou
* @createTime 2020年05月29日 11:34:00
*/
@Slf4j
public
class
NettyClientTransport
implements
ClientTransport
{
private
final
ServiceDiscovery
serviceDiscovery
;
private
final
UnprocessedRequests
unprocessedRequests
;
private
final
ChannelProvider
channelProvider
;
public
NettyClientTransport
()
{
this
.
serviceDiscovery
=
ExtensionLoader
.
getExtensionLoader
(
ServiceDiscovery
.
class
).
getExtension
(
"zk"
);
this
.
unprocessedRequests
=
SingletonFactory
.
getInstance
(
UnprocessedRequests
.
class
);
this
.
channelProvider
=
SingletonFactory
.
getInstance
(
ChannelProvider
.
class
);
}
@Override
public
CompletableFuture
<
RpcResponse
<
Object
>>
sendRpcRequest
(
RpcRequest
rpcRequest
)
{
// build return value
CompletableFuture
<
RpcResponse
<
Object
>>
resultFuture
=
new
CompletableFuture
<>();
// build rpc service name by rpcRequest
String
rpcServiceName
=
rpcRequest
.
toRpcProperties
().
toRpcServiceName
();
// get server address
InetSocketAddress
inetSocketAddress
=
serviceDiscovery
.
lookupService
(
rpcServiceName
);
// get server address related channel
Channel
channel
=
channelProvider
.
get
(
inetSocketAddress
);
if
(
channel
!=
null
&&
channel
.
isActive
())
{
// put unprocessed request
unprocessedRequests
.
put
(
rpcRequest
.
getRequestId
(),
resultFuture
);
RpcMessage
rpcMessage
=
new
RpcMessage
();
rpcMessage
.
setData
(
rpcRequest
);
rpcMessage
.
setCodec
(
SerializationTypeEnum
.
PROTOSTUFF
.
getCode
());
rpcMessage
.
setCompress
(
CompressTypeEnum
.
GZIP
.
getCode
());
rpcMessage
.
setMessageType
(
RpcConstants
.
REQUEST_TYPE
);
channel
.
writeAndFlush
(
rpcMessage
).
addListener
((
ChannelFutureListener
)
future
->
{
if
(
future
.
isSuccess
())
{
log
.
info
(
"client send message: [{}]"
,
rpcMessage
);
}
else
{
future
.
channel
().
close
();
resultFuture
.
completeExceptionally
(
future
.
cause
());
log
.
error
(
"Send failed:"
,
future
.
cause
());
}
});
}
else
{
throw
new
IllegalStateException
();
}
return
resultFuture
;
}
}
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyClient.java
→
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/Netty
Rpc
Client.java
浏览文件 @
5959b8ee
package
github.javaguide.remoting.transport.netty.client
;
import
github.javaguide.enums.CompressTypeEnum
;
import
github.javaguide.enums.SerializationTypeEnum
;
import
github.javaguide.extension.ExtensionLoader
;
import
github.javaguide.factory.SingletonFactory
;
import
github.javaguide.registry.ServiceDiscovery
;
import
github.javaguide.remoting.constants.RpcConstants
;
import
github.javaguide.remoting.dto.RpcMessage
;
import
github.javaguide.remoting.dto.RpcRequest
;
import
github.javaguide.remoting.dto.RpcResponse
;
import
github.javaguide.remoting.transport.RpcRequestTransport
;
import
github.javaguide.remoting.transport.netty.codec.RpcMessageDecoder
;
import
github.javaguide.remoting.transport.netty.codec.RpcMessageEncoder
;
import
io.netty.bootstrap.Bootstrap
;
...
...
@@ -30,12 +40,15 @@ import java.util.concurrent.TimeUnit;
* @createTime 2020年05月29日 17:51:00
*/
@Slf4j
public
final
class
NettyClient
{
public
final
class
NettyRpcClient
implements
RpcRequestTransport
{
private
final
ServiceDiscovery
serviceDiscovery
;
private
final
UnprocessedRequests
unprocessedRequests
;
private
final
ChannelProvider
channelProvider
;
private
final
Bootstrap
bootstrap
;
private
final
EventLoopGroup
eventLoopGroup
;
// initialize resources such as EventLoopGroup, Bootstrap
public
NettyClient
()
{
public
NettyRpcClient
()
{
// initialize resources such as EventLoopGroup, Bootstrap
eventLoopGroup
=
new
NioEventLoopGroup
();
bootstrap
=
new
Bootstrap
();
bootstrap
.
group
(
eventLoopGroup
)
...
...
@@ -52,12 +65,14 @@ public final class NettyClient {
p
.
addLast
(
new
IdleStateHandler
(
0
,
5
,
0
,
TimeUnit
.
SECONDS
));
p
.
addLast
(
new
RpcMessageEncoder
());
p
.
addLast
(
new
RpcMessageDecoder
());
p
.
addLast
(
new
NettyClientHandler
());
p
.
addLast
(
new
Netty
Rpc
ClientHandler
());
}
});
this
.
serviceDiscovery
=
ExtensionLoader
.
getExtensionLoader
(
ServiceDiscovery
.
class
).
getExtension
(
"zk"
);
this
.
unprocessedRequests
=
SingletonFactory
.
getInstance
(
UnprocessedRequests
.
class
);
this
.
channelProvider
=
SingletonFactory
.
getInstance
(
ChannelProvider
.
class
);
}
/**
* connect server and get the channel ,so that you can send rpc message to server
*
...
...
@@ -78,8 +93,50 @@ public final class NettyClient {
return
completableFuture
.
get
();
}
@Override
public
Object
sendRpcRequest
(
RpcRequest
rpcRequest
)
{
// build return value
CompletableFuture
<
RpcResponse
<
Object
>>
resultFuture
=
new
CompletableFuture
<>();
// build rpc service name by rpcRequest
String
rpcServiceName
=
rpcRequest
.
toRpcProperties
().
toRpcServiceName
();
// get server address
InetSocketAddress
inetSocketAddress
=
serviceDiscovery
.
lookupService
(
rpcServiceName
);
// get server address related channel
Channel
channel
=
getChannel
(
inetSocketAddress
);
if
(
channel
.
isActive
())
{
// put unprocessed request
unprocessedRequests
.
put
(
rpcRequest
.
getRequestId
(),
resultFuture
);
RpcMessage
rpcMessage
=
new
RpcMessage
();
rpcMessage
.
setData
(
rpcRequest
);
rpcMessage
.
setCodec
(
SerializationTypeEnum
.
PROTOSTUFF
.
getCode
());
rpcMessage
.
setCompress
(
CompressTypeEnum
.
GZIP
.
getCode
());
rpcMessage
.
setMessageType
(
RpcConstants
.
REQUEST_TYPE
);
channel
.
writeAndFlush
(
rpcMessage
).
addListener
((
ChannelFutureListener
)
future
->
{
if
(
future
.
isSuccess
())
{
log
.
info
(
"client send message: [{}]"
,
rpcMessage
);
}
else
{
future
.
channel
().
close
();
resultFuture
.
completeExceptionally
(
future
.
cause
());
log
.
error
(
"Send failed:"
,
future
.
cause
());
}
});
}
else
{
throw
new
IllegalStateException
();
}
return
resultFuture
;
}
public
Channel
getChannel
(
InetSocketAddress
inetSocketAddress
)
{
Channel
channel
=
channelProvider
.
get
(
inetSocketAddress
);
if
(
channel
==
null
)
{
channel
=
doConnect
(
inetSocketAddress
);
channelProvider
.
set
(
inetSocketAddress
,
channel
);
}
return
channel
;
}
public
void
close
()
{
eventLoopGroup
.
shutdownGracefully
();
}
}
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyClientHandler.java
→
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/Netty
Rpc
ClientHandler.java
浏览文件 @
5959b8ee
...
...
@@ -30,13 +30,13 @@ import java.net.InetSocketAddress;
* @createTime 2020年05月25日 20:50:00
*/
@Slf4j
public
class
NettyClientHandler
extends
ChannelInboundHandlerAdapter
{
public
class
Netty
Rpc
ClientHandler
extends
ChannelInboundHandlerAdapter
{
private
final
UnprocessedRequests
unprocessedRequests
;
private
final
ChannelProvider
channelProvider
;
private
final
NettyRpcClient
nettyRpcClient
;
public
NettyClientHandler
()
{
public
Netty
Rpc
ClientHandler
()
{
this
.
unprocessedRequests
=
SingletonFactory
.
getInstance
(
UnprocessedRequests
.
class
);
this
.
channelProvider
=
SingletonFactory
.
getInstance
(
ChannelProvider
.
class
);
this
.
nettyRpcClient
=
SingletonFactory
.
getInstance
(
NettyRpcClient
.
class
);
}
/**
...
...
@@ -67,7 +67,7 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
IdleState
state
=
((
IdleStateEvent
)
evt
).
state
();
if
(
state
==
IdleState
.
WRITER_IDLE
)
{
log
.
info
(
"write idle happen [{}]"
,
ctx
.
channel
().
remoteAddress
());
Channel
channel
=
channelProvider
.
get
((
InetSocketAddress
)
ctx
.
channel
().
remoteAddress
());
Channel
channel
=
nettyRpcClient
.
getChannel
((
InetSocketAddress
)
ctx
.
channel
().
remoteAddress
());
RpcMessage
rpcMessage
=
new
RpcMessage
();
rpcMessage
.
setCodec
(
SerializationTypeEnum
.
PROTOSTUFF
.
getCode
());
rpcMessage
.
setCompress
(
CompressTypeEnum
.
GZIP
.
getCode
());
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/server/NettyServer.java
→
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/server/Netty
Rpc
Server.java
浏览文件 @
5959b8ee
...
...
@@ -38,7 +38,7 @@ import java.util.concurrent.TimeUnit;
*/
@Slf4j
@Component
public
class
NettyServer
{
public
class
Netty
Rpc
Server
{
public
static
final
int
PORT
=
9998
;
...
...
@@ -78,7 +78,7 @@ public class NettyServer {
p
.
addLast
(
new
IdleStateHandler
(
30
,
0
,
0
,
TimeUnit
.
SECONDS
));
p
.
addLast
(
new
RpcMessageEncoder
());
p
.
addLast
(
new
RpcMessageDecoder
());
p
.
addLast
(
serviceHandlerGroup
,
new
NettyServerHandler
());
p
.
addLast
(
serviceHandlerGroup
,
new
Netty
Rpc
ServerHandler
());
}
});
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/server/NettyServerHandler.java
→
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/server/Netty
Rpc
ServerHandler.java
浏览文件 @
5959b8ee
...
...
@@ -28,11 +28,11 @@ import lombok.extern.slf4j.Slf4j;
* @createTime 2020年05月25日 20:44:00
*/
@Slf4j
public
class
NettyServerHandler
extends
ChannelInboundHandlerAdapter
{
public
class
Netty
Rpc
ServerHandler
extends
ChannelInboundHandlerAdapter
{
private
final
RpcRequestHandler
rpcRequestHandler
;
public
NettyServerHandler
()
{
public
Netty
Rpc
ServerHandler
()
{
this
.
rpcRequestHandler
=
SingletonFactory
.
getInstance
(
RpcRequestHandler
.
class
);
}
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/socket/SocketRpcClient.java
浏览文件 @
5959b8ee
...
...
@@ -5,7 +5,7 @@ import github.javaguide.exception.RpcException;
import
github.javaguide.extension.ExtensionLoader
;
import
github.javaguide.registry.ServiceDiscovery
;
import
github.javaguide.remoting.dto.RpcRequest
;
import
github.javaguide.remoting.transport.
Clien
tTransport
;
import
github.javaguide.remoting.transport.
RpcReques
tTransport
;
import
lombok.AllArgsConstructor
;
import
lombok.extern.slf4j.Slf4j
;
...
...
@@ -23,7 +23,7 @@ import java.net.Socket;
*/
@AllArgsConstructor
@Slf4j
public
class
SocketRpcClient
implements
Clien
tTransport
{
public
class
SocketRpcClient
implements
RpcReques
tTransport
{
private
final
ServiceDiscovery
serviceDiscovery
;
public
SocketRpcClient
()
{
...
...
rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/socket/SocketRpcServer.java
浏览文件 @
5959b8ee
...
...
@@ -15,7 +15,7 @@ import java.net.ServerSocket;
import
java.net.Socket
;
import
java.util.concurrent.ExecutorService
;
import
static
github
.
javaguide
.
remoting
.
transport
.
netty
.
server
.
NettyServer
.
PORT
;
import
static
github
.
javaguide
.
remoting
.
transport
.
netty
.
server
.
Netty
Rpc
Server
.
PORT
;
/**
* @author shuang.kou
...
...
@@ -34,7 +34,6 @@ public class SocketRpcServer {
serviceProvider
=
SingletonFactory
.
getInstance
(
ServiceProviderImpl
.
class
);
}
public
void
registerService
(
Object
service
)
{
serviceProvider
.
publishService
(
service
);
}
...
...
rpc-framework-simple/src/main/java/github/javaguide/spring/SpringBeanPostProcessor.java
浏览文件 @
5959b8ee
...
...
@@ -8,7 +8,7 @@ import github.javaguide.factory.SingletonFactory;
import
github.javaguide.provider.ServiceProvider
;
import
github.javaguide.provider.ServiceProviderImpl
;
import
github.javaguide.proxy.RpcClientProxy
;
import
github.javaguide.remoting.transport.
Clien
tTransport
;
import
github.javaguide.remoting.transport.
RpcReques
tTransport
;
import
lombok.SneakyThrows
;
import
lombok.extern.slf4j.Slf4j
;
import
org.springframework.beans.BeansException
;
...
...
@@ -28,11 +28,11 @@ import java.lang.reflect.Field;
public
class
SpringBeanPostProcessor
implements
BeanPostProcessor
{
private
final
ServiceProvider
serviceProvider
;
private
final
Clien
tTransport
rpcClient
;
private
final
RpcReques
tTransport
rpcClient
;
public
SpringBeanPostProcessor
()
{
this
.
serviceProvider
=
SingletonFactory
.
getInstance
(
ServiceProviderImpl
.
class
);
this
.
rpcClient
=
ExtensionLoader
.
getExtensionLoader
(
ClientTransport
.
class
).
getExtension
(
"nettyClientTransport
"
);
this
.
rpcClient
=
ExtensionLoader
.
getExtensionLoader
(
RpcRequestTransport
.
class
).
getExtension
(
"netty
"
);
}
@SneakyThrows
...
...
rpc-framework-simple/src/main/resources/META-INF/extensions/github.javaguide.remoting.transport.ClientTransport
已删除
100644 → 0
浏览文件 @
801287f4
nettyClientTransport=github.javaguide.remoting.transport.netty.client.NettyClientTransport
socketRpcClient=github.javaguide.remoting.transport.socket.SocketRpcClient
\ No newline at end of file
rpc-framework-simple/src/main/resources/META-INF/extensions/github.javaguide.remoting.transport.RpcRequestTransport
0 → 100644
浏览文件 @
5959b8ee
netty=github.javaguide.remoting.transport.netty.client.NettyRpcClient
socket=github.javaguide.remoting.transport.socket.SocketRpcClient
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录