Skip to content

Latest commit

 

History

History
282 lines (226 loc) · 8.79 KB

what's-microservice-how-to-communicate.md

File metadata and controls

282 lines (226 loc) · 8.79 KB

什么是微服务？微服务之间是如何独立通讯的？

什么是微服务

  • 微服务架构是一个分布式系统，按照业务进行划分成为不同的服务单元，解决单体系统性能等不足。
  • 微服务是一种架构风格，一个大型软件应用由多个服务单元组成。系统中的服务单元可以单独部署，各个服务单元之间是松耦合的。

微服务概念起源：Microservices

微服务之间是如何独立通讯的

同步

REST HTTP 协议

REST 请求在微服务中是最为常用的一种通讯方式，它依赖于 HTTP\HTTPS 协议。RESTFUL 的特点是：

  1. 每一个 URI 代表 1 种资源
  2. 客户端使用 GET、POST、PUT、DELETE 4 个表示操作方式的动词对服务端资源进行操作：GET 用来获取资源，POST 用来新建资源（也可以用于更新资源），PUT 用来更新资源，DELETE 用来删除资源
  3. 通过操作资源的表现形式来操作资源
  4. 资源的表现形式是 XML 或者 HTML
  5. 客户端与服务端之间的交互在请求之间是无状态的,从客户端到服务端的每个请求都必须包含理解请求所必需的信息

举个例子，有一个服务方提供了如下接口：

@RestController@RequestMapping("/communication") publicclassRestControllerDemo { @GetMapping("/hello") publicStrings() { return"hello"; } }

另外一个服务需要去调用该接口，调用方只需要根据 API 文档发送请求即可获取返回结果。

@RestController@RequestMapping("/demo") publicclassRestDemo{ @AutowiredRestTemplaterestTemplate; @GetMapping("/hello2") publicStrings2() { StringforObject = restTemplate.getForObject("http://localhost:9013/communication/hello", String.class); returnforObject; } }

通过这样的方式可以实现服务之间的通讯。

RPC TCP 协议

RPC(Remote Procedure Call)远程过程调用，简单的理解是一个节点请求另一个节点提供的服务。它的工作流程是这样的：

  1. 执行客户端调用语句，传送参数
  2. 调用本地系统发送网络消息
  3. 消息传送到远程主机
  4. 服务器得到消息并取得参数
  5. 根据调用请求以及参数执行远程过程（服务）
  6. 执行过程完毕，将结果返回服务器句柄
  7. 服务器句柄返回结果，调用远程主机的系统网络服务发送结果
  8. 消息传回本地主机
  9. 客户端句柄由本地主机的网络服务接收消息
  10. 客户端接收到调用语句返回的结果数据

举个例子。

首先需要一个服务端：

importjava.io.IOException; importjava.io.ObjectInputStream; importjava.io.ObjectOutputStream; importjava.lang.reflect.Method; importjava.net.InetSocketAddress; importjava.net.ServerSocket; importjava.net.Socket; importjava.util.concurrent.ConcurrentHashMap; importjava.util.concurrent.ExecutorService; importjava.util.concurrent.Executors; /** * RPC 服务端用来注册远程方法的接口和实现类 */publicclassRPCServer { privatestaticExecutorServiceexecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); privatestaticfinalConcurrentHashMap<String, Class> serviceRegister = newConcurrentHashMap<>(); /** * 注册方法 * @param service * @param impl */publicvoidregister(Classservice, Classimpl) { serviceRegister.put(service.getSimpleName(), impl); } /** * 启动方法 * @param port */publicvoidstart(intport) { ServerSocketsocket = null; try { socket = newServerSocket(); socket.bind(newInetSocketAddress(port)); System.out.println("服务启动"); System.out.println(serviceRegister); while (true) { executor.execute(newTask(socket.accept())); } } catch (Exceptione) { e.printStackTrace(); } finally { if (socket != null) { try { socket.close(); } catch (IOExceptione) { e.printStackTrace(); } } } } privatestaticclassTaskimplementsRunnable { Socketclient = null; publicTask(Socketclient) { this.client = client; } @Overridepublicvoidrun() { ObjectInputStreaminput = null; ObjectOutputStreamoutput = null; try { input = newObjectInputStream(client.getInputStream()); // 按照顺序读取对方写过来的内容StringserviceName = input.readUTF(); StringmethodName = input.readUTF(); Class<?>[] parameterTypes = (Class<?>[]) input.readObject(); Object[] arguments = (Object[]) input.readObject(); ClassserviceClass = serviceRegister.get(serviceName); if (serviceClass == null) { thrownewClassNotFoundException(serviceName + " 没有找到!"); } Methodmethod = serviceClass.getMethod(methodName, parameterTypes); Objectresult = method.invoke(serviceClass.newInstance(), arguments); output = newObjectOutputStream(client.getOutputStream()); output.writeObject(result); } catch (Exceptione) { e.printStackTrace(); } finally { try { // 这里就不写 output!=null才关闭这个逻辑了output.close(); input.close(); client.close(); } catch (IOExceptione) { e.printStackTrace(); } } } } }

其次需要一个客户端：

importjava.io.ObjectInputStream; importjava.io.ObjectOutputStream; importjava.lang.reflect.InvocationHandler; importjava.lang.reflect.Method; importjava.lang.reflect.Proxy; importjava.net.InetSocketAddress; importjava.net.Socket; /** * RPC 客户端 */publicclassRPCclient<T> { /** * 通过动态代理将参数发送过去到 RPCServer ,RPCserver 返回结果这个方法处理成为正确的实体 */publicstatic <T> TgetRemoteProxyObj(finalClass<T> service, finalInetSocketAddressaddr) { return (T) Proxy.newProxyInstance(service.getClassLoader(), newClass<?>[]{service}, newInvocationHandler() { @OverridepublicObjectinvoke(Objectproxy, Methodmethod, Object[] args) throwsThrowable { Socketsocket = null; ObjectOutputStreamout = null; ObjectInputStreaminput = null; try { socket = newSocket(); socket.connect(addr); // 将实体类,参数,发送给远程调用方out = newObjectOutputStream(socket.getOutputStream()); out.writeUTF(service.getSimpleName()); out.writeUTF(method.getName()); out.writeObject(method.getParameterTypes()); out.writeObject(args); input = newObjectInputStream(socket.getInputStream()); returninput.readObject(); } catch (Exceptione) { e.printStackTrace(); } finally { out.close(); input.close(); socket.close(); } returnnull; } }); } }

再来一个测试的远程方法。

publicinterfaceTinterface { Stringsend(Stringmsg); } publicclassTinterfaceImplimplementsTinterface { @OverridepublicStringsend(Stringmsg) { return"send message " + msg; } }

测试代码如下：

importjava.net.InetSocketAddress; publicclassRunTest { publicstaticvoidmain(String[] args) { newThread(newRunnable() { @Overridepublicvoidrun() { RPCServerrpcServer = newRPCServer(); rpcServer.register(Tinterface.class, TinterfaceImpl.class); rpcServer.start(10000); } }).start(); Tinterfacetinterface = RPCclient.getRemoteProxyObj(Tinterface.class, newInetSocketAddress("localhost", 10000)); System.out.println(tinterface.send("rpc 测试用例")); } }

输出 send message rpc 测试用例 。

异步

消息中间件

常见的消息中间件有 Kafka、ActiveMQ、RabbitMQ、RocketMQ ，常见的协议有 AMQP、MQTTP、STOMP、XMPP。这里不对消息队列进行拓展了，具体如何使用还是请移步官网。

close