掃二維碼與項目經理溝通
我們在微信上24小時期待你的聲音
解答本文疑問/技術咨詢/運營咨詢/技術建議/互聯(lián)網交流
本文的宗旨在于通過簡單干凈實踐的方式教會讀者，為什么要使用Dubbo、怎么使用Dubbo、Dubbo通信的原理是什么。在學習本文后，你可以避開很多關于 Dubbo 使用時的坑，也能更清楚自己的編碼是在做什么。
成都一家集口碑和實力的網站建設服務商，擁有專業(yè)的企業(yè)建站團隊和靠譜的建站技術，十多年企業(yè)及個人網站建設經驗 ,為成都成百上千家客戶提供網頁設計制作,網站開發(fā),企業(yè)網站制作建設等服務,包括成都營銷型網站建設，品牌網站制作，同時也為不同行業(yè)的客戶提供網站設計制作、網站制作的服務,包括成都電商型網站制作建設,裝修行業(yè)網站制作建設，傳統(tǒng)機械行業(yè)網站建設,傳統(tǒng)農業(yè)行業(yè)網站制作建設。在成都做網站,選網站制作建設服務商就選創(chuàng)新互聯(lián)。
本文涉及的工程：
隨著互聯(lián)網場景中所要面對的用戶規(guī)模和體量的增加，系統(tǒng)的也需要做相應的拆分設計和實現(xiàn)。隨之而來的，以前的一套系統(tǒng)，現(xiàn)在成了多個微服務。如；電商系統(tǒng)，以前就在一個工程中寫就可以了，現(xiàn)在需要拆分出，用戶、支付、商品、配送、活動、風控等各個模塊。那么這些模塊拆分后，如何高效的通信呢？
圖片
圖片
Dubbo 的使用分為2方，一個是接口的提供方，另外一個是接口的調用方。接口的提供方需要提供出被調用方使用接口的描述性信息。這個信息包括；接口名稱、接口入參、接口出參，只有讓調用方拿到這些信息以后，它才能依托于這樣的接口信息做一個代理操作，并在代理類中使用 Socket 完成雙方的信息交互。
所以你看上去調用 RPC 接口好像和使用 HTTP 也沒啥區(qū)別，無非就是引入了 POM 配置，之后再配置了注解就可以使用了。但其實，它是把你的 Jar 當做代理的必要參數使用了。本文也會介紹，具體是怎么代理的
對于編程的學習來說，其實最開始的那一下，不是搞明白所有原理，而是先讓自己可以看到運行出來的效果。哎，之后就去分析原理，這樣會舒服的多。
所以小傅哥這里提供了一套簡單的 Dubbo 使用案例，只要你滿足最基本的配置條件，就可以運行出效果；
工程案例創(chuàng)建結構，采用的是 DDD 結構。但和 DDD 一點關系沒有。如果你對工程創(chuàng)建有疑惑，可以參考 《Java 簡明教程》之 DDD 架構
圖片
源碼：cn.bugstack.dev.tech.dubbo.api.IUserService
public interface IUserService {
Response queryUserInfo(UserReqDTO reqDTO);
} 源碼：cn.bugstack.dev.tech.dubbo.trigger.rpc.UserService
@Slf4j
@DubboService(version = "1.0.0")
public class UserService implements IUserService {
@Override
public Response queryUserInfo(UserReqDTO reqDTO) {
log.info("查詢用戶信息 userId: {} reqStr: {}", reqDTO.getUserId(), JSON.toJSONString(reqDTO));
try {
// 1. 模擬查詢【你可以從數據庫或者Redis緩存獲取數據】
UserResDTO resDTO = UserResDTO.builder()
.userId(reqDTO.getUserId())
.userName("小傅哥")
.userAge(20)
.build();
// 2. 返回結果
return Response.builder()
.code(Constants.ResponseCode.SUCCESS.getCode())
.info(Constants.ResponseCode.SUCCESS.getInfo())
.data(resDTO).build();
} catch (Exception e) {
log.error("查詢用戶信息失敗 userId: {} reqStr: {}", reqDTO.getUserId(), JSON.toJSONString(reqDTO), e);
return Response.builder()
.code(Constants.ResponseCode.UN_ERROR.getCode())
.info(Constants.ResponseCode.UN_ERROR.getInfo())
.build();
}
}
} application.yml
dubbo:
application:
name: xfg-dev-tech-dubbo
version: 1.0.0
registry:
address: zookeeper://127.0.0.1:2181 # N/A - 無zookeeper可配置 N/A 走直連模式測試
protocol:
name: dubbo
port: 20881
scan:
base-packages: cn.bugstack.dev.tech.dubbo.api以上信息都準備了，一群小卡拉米開始掉到第4個坑里了！
你有2個應用，一個Dubbo接口提供方、一個Dubbo接口使用方。那么你在給你另外一個應用使用接口的時候，你在 InelliJ IDEA 的 Maven 中執(zhí)行 Install 了嗎？
Install 是干啥的？它是為了讓你使用了同一個本地 Maven 配置的應用，可以引入到對方提供的 Jar 包。你 Install 以后，這個 Jar 包就會進入到本地 Maven 倉庫了。如果是公司里開發(fā)，會有專門的自己家部署的，私有Maven中心倉庫，就可以通過 deploy 把本地 Jar 發(fā)布上去，那么公司里的伙伴，也就都可以引用了。
圖片
有些小卡拉米覺得前面的抗都掃干凈了，就完事了。沒有接下來還有坑，讓你一搞搞一天，半夜也睡不好。
cn.bugstack
xfg-dev-tech-dubbo-api
1.0-SNAPSHOT
源碼：application.yml
dubbo:
application:
name: xfg-dev-tech-dubbo
version: 1.0.0
registry:
address: zookeeper://127.0.0.1:2181
# address: N/A
protocol:
name: dubbo
port: 20881源碼：cn.bugstack.dev.tech.dubbo.consumer.test.ApiTest
// 直連模式；@DubboReference(interfaceClass = IUserService.class, url = "dubbo://127.0.0.1:20881", version = "1.0.0")
@DubboReference(interfaceClass = IUserService.class, version = "1.0.0")
private IUserService userService;
@Test
public void test_userService() {
UserReqDTO reqDTO = UserReqDTO.builder().userId("10001").build();
Response resDTO = userService.queryUserInfo(reqDTO);
log.info("測試結果 req: {} res: {}", JSON.toJSONString(reqDTO), JSON.toJSONString(resDTO));
} 測試結果
2023-07-08 15:37:22.291 INFO 62481 --- [ main] c.b.d.tech.dubbo.consumer.test.ApiTest : 測試結果 req: {"userId":"10001"} res: {"code":"0000","data":{"userAge":20,"userId":"10001","userName":"小傅哥"},"info":"成功"}
2023-07-08 15:37:22.324 INFO 62481 --- [tor-Framework-0] o.a.c.f.imps.CuratorFrameworkImpl : backgroundOperationsLoop exiting都說 Jar 是提供可描述性信息的，對方才能代理調用。那么這個過程是怎么干的呢，總不能一問這個，就讓小卡拉米們去手寫 Dubbo 呀！所以小傅哥會通過最簡單模型結構，讓你了解這個 Dubbo 通信的原理，方便小卡拉米們上手。
圖片
好，核心的原理就這么點。接下來，我們從代碼中看看。
源碼：cn.bugstack.dev.tech.dubbo.trigger.socket.RpcServerSocket
@Slf4j
@Service
public class RpcServerSocket implements Runnable {
private ApplicationContext applicationContext;
public RpcServerSocket(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
new Thread(this).start();
}
@Override
public void run() {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childHandler(new ChannelInitializer() {
@Override
public void initChannel(SocketChannel channel) {
channel.pipeline().addLast(new ObjectEncoder());
channel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
channel.pipeline().addLast(new SimpleChannelInboundHandler>() {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Map request) throws Exception {
// 解析參數
Class clazz = (Class) request.get("clazz");
String methodName = (String) request.get("methodName");
Class[] paramTypes = (Class[]) request.get("paramTypes");
Object[] args = (Object[]) request.get("args");
// 反射調用
Method method = clazz.getMethod(methodName, paramTypes);
Object invoke = method.invoke(applicationContext.getBean(clazz), args);
// 封裝結果
Map response = new HashMap<>();
response.put("data", invoke);
log.info("RPC 請求調用 clazz:{} methodName:{}, response:{}", clazz.getName(), methodName, JSON.toJSON(response));
// 回寫數據
channelHandlerContext.channel().writeAndFlush(response);
}
});
}
});
ChannelFuture f = b.bind(22881).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
} 這段代碼主要提供的功能包括；
打開工程：xfg-dev-tech-dubbo-test
源碼：cn.bugstack.dev.tech.dubbo.consumer.config.RPCProxyBeanFactory
@Slf4j
@Component("rpcProxyBeanFactory")
public class RPCProxyBeanFactory implements FactoryBean, Runnable {
private Channel channel;
// 緩存數據，實際RPC會對每次的調用生成一個ID來標記獲取
private Object responseCache;
public RPCProxyBeanFactory() throws InterruptedException {
new Thread(this).start();
while (null == channel) {
Thread.sleep(150);
log.info("Rpc Socket 鏈接等待...");
}
}
@Override
public IUserService getObject() throws Exception {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Class[] classes = {IUserService.class};
InvocationHandler handler = new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
if (Object.class.equals(method.getDeclaringClass())) {
return method.invoke(this, args);
}
Map request = new HashMap<>();
request.put("clazz", IUserService.class);
request.put("methodName", method.getName());
request.put("paramTypes", method.getParameterTypes());
request.put("args", args);
channel.writeAndFlush(request);
// 模擬超時等待，一般RPC接口請求，都有一個超時等待時長。
Thread.sleep(350);
return responseCache;
}
};
return (IUserService) Proxy.newProxyInstance(classLoader, classes, handler);
}
@Override
public Class getObjectType() {
return IUserService.class;
}
@Override
public void run() {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.AUTO_READ, true)
.handler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(new ObjectEncoder());
channel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
channel.pipeline().addLast(new SimpleChannelInboundHandler>() {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Map data) throws Exception {
responseCache = data.get("data");
}
});
}
});
ChannelFuture channelFuture = b.connect("127.0.0.1", 22881).syncUninterruptibly();
this.channel = channelFuture.channel();
channelFuture.channel().closeFuture().syncUninterruptibly();
} finally {
workerGroup.shutdownGracefully();
}
}
} 這段代碼主要提供的功能包括；
@Resource(name = "rpcProxyBeanFactory")
private IUserService proxyUserService;
@Test
public void test_proxyUserService(){
UserReqDTO reqDTO = UserReqDTO.builder().userId("10001").build();
Response resDTO = proxyUserService.queryUserInfo(reqDTO);
log.info("測試結果 req: {} res: {}", JSON.toJSONString(reqDTO), JSON.toJSONString(resDTO));
} 測試結果
2023-07-08 16:14:51.322 INFO 74498 --- [ main] c.b.d.tech.dubbo.consumer.test.ApiTest : 測試結果 req: {"userId":"10001"} res: {"code":"0000","data":{"userAge":20,"userId":"10001","userName":"小傅哥"},"info":"成功"}
我們在微信上24小時期待你的聲音
解答本文疑問/技術咨詢/運營咨詢/技術建議/互聯(lián)網交流
微信二維碼
移動版官網