網路編程3-有解無憂

網路通信基本常識

編程中的Socket是什么？
- Socket是應用層與TCP/IP協議族通信的間軟體抽象層，它是一組介面，其實就是一個門面模式，TCP用主機的IP地址加上主機上的埠號作為TCP連接的端點，這種端點就叫做套接字（socket），
- socket是一種門面模式，屏蔽了與機器的交流程序，方便業務程式員開發代碼
- A的socket ---- B的socket
  
  兩個socket建立關系，從而A和B就可以通信了
短連接
- 一次通訊完了馬上就關閉連接
- 服務器要服務成千上億的服務
- 短連接和長連接到底如何選擇，需要根據實際情況
長連接
- 一次連接內會進行多次通訊
- 維持住長連接主要消耗的是記憶體和FD檔案描述符
- 如果服務器性能好，可以維持上百萬的長連接
- 操作頻繁、或者點對點的通訊上，連接池來維護這種長連接
InetAddress類 --> 只表示地址(主機)
- InetAddress address = InetAddress.getByName(“www.baidu.com”)
- 可以獲得百度的ip地址
- 怎么獲得的？-- dns決議
- InetAddress address2 = InetAddress.getByName(“124.232.170.22”)
  
  System.out.println(address2.getHostName())
  
  同樣可以根據ip地址獲取域名，如果找不到，就會列印源ip地址
- InetAddress[] allIp= InetAddress.getAllByName(“www.baidu.com”)
  
  獲取百度域名下面所有的ip地址
- 把一個ip地址包裝成InetAddress物件
  
  byte[] bytes = {(byte)192,(byte)168,56,1};
  
  InetAddress address = InetAddress.getByAddress(bytes);
InetSocketAddress類 --> 主機名+埠

NetoworkInterface類
- 打開設備管理器-網路配接器
  
  NetoworkInterface會把這些全部找出來，除了這些，還有別的
  
  // 127.0.0.1-------本地回環介面
  
  InetAddress address = InetAddress.getByName(“127.0.0.1”);
  
  NetworkInterface byInetAddress = NetoworkInterface.getByInetAddress(address);
  
  // 獲取所有的網路通訊介面類
  
  // 包括本地回環、網路配接器、wifi、廣域網、硬體設備通訊
  
  NetoworkInterface.getNetoworkInterfaces();
服務端、客戶端、通信編程關注的三件事
- 所有網路通訊一定有服務端和客戶端這兩樣
- 提供服務的稱為服務端
- 連接服務的稱為客戶端
- 某個類有Server、ServerSocket，那么這個類往往是給服務端使用的
  
  ServerSocket只是個容器，容納網路服務用的
- 某個類只有socket，一般是負責具體的網路讀寫
  
  真實網路通訊中，真正進行網路讀寫的都是socket
網路編程中一定要關注的點
- 網路連接
- 讀網路資料
- 寫網路資料
jdk網路通訊
- bio
- nio
- aio—非主流

原生JDK網路編程-BIO

BIO：block ，即阻塞式io
ServerSocket負責系結IP地址，啟動監聽埠；
Socket負責發起連接操作，連接成功后，雙方通過輸入和輸出流進行同步阻塞式通信

ServerSocket啟動了處于監聽狀態

通過accept()方法來回應請求的socket()

新生成一個執行緒，在執行緒中生成一個socket()來與請求的socket()通訊

服務器端

package WangLuoBianCheng.wangLuoBianCheng3;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;

public class Server{

    public static void main(String[] args)throws IOException {

        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress(10001));
        System.out.println("Server is started...");
        while(true){
            // 來一個socket請求，就會new一個執行緒，同時new 一個socket
            // 執行server.accept()成功，如果三次握手完成會回傳一個Socket
            // 更優的寫法是用執行緒池，實作復用
            new Thread(new ServerTask(server.accept())).start();

        }

    }


    private static class ServerTask implements Runnable{

        private Socket socket = null;

        public ServerTask(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run(){
            try (ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                 ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream())){

                String userName = input.readUTF();
                System.out.println("Accept client Message:"+userName);

                output.writeUTF("Hello,"+userName);
                // 上一句只是寫入快取，準備發送
                // 強制刷出
                output.flush();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

}

客戶端

package WangLuoBianCheng.wangLuoBianCheng3;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;

public class Client {

    public static void main(String[] args) throws IOException {
        // 用來通訊的socket
        Socket socket = null;

        // 注意客戶端必須是輸出在上，輸入在下，確保流通道建立起來
        // 因為如果兩端都是輸入，怎么傳遞？
        // 或者兩端都是先創建輸出流，再創建輸入流，未測驗？
        ObjectOutputStream objectOutputStream = null;
        InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 10001);
        ObjectInputStream objectInputStream = null;

        try {
            socket = new Socket();
            // 建立連接
            socket.connect(inetSocketAddress);

            objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
            objectInputStream = new ObjectInputStream(socket.getInputStream());

            // 發出訊息
            objectOutputStream.writeUTF("James");
            objectOutputStream.flush();

            // 接受服務器回應的資訊并列印
            // 執行完后關閉了網路連接
            System.out.println(objectInputStream.readUTF());

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (socket!=null) socket.close();

            if (objectOutputStream!=null) objectOutputStream.close();

            if (objectInputStream!=null) objectInputStream.close();
        }
    }
}

改進—偽異步IO模型

上述服務器端是來一個socket請求，就會new一個執行緒，同時new 一個socket
為了實作執行緒復用，應該采用執行緒池的執行緒來處理任務，而這種模式又稱為偽異步IO模型

package WangLuoBianCheng.wangLuoBianCheng3;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ServerPool {

    private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()*2);


    public static void main(String[] args)throws IOException {

        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress(10001));
        System.out.println("Server is started...");
        while(true){
            // 執行server.accept()成功，如果三次握手完成會回傳一個Socket
            executorService.execute(new ServerTask(server.accept()));
        }

    }


    private static class ServerTask implements Runnable{

        private Socket socket = null;

        public ServerTask(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run(){
            try (ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                 ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream())){

                String userName = input.readUTF();
                System.out.println("Accept client Message:"+userName);

                output.writeUTF("Hello,"+userName);
                // 強制刷出
                output.flush();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}

BIO應用-RPC框架

單WEB專案
- 1.訂單服務
- 2.扣減庫存服務
- 3.發送短信提示服務
之前上述三個服務全是單執行緒執行，當呼叫服務的用戶數很大時，就不能充分發揮并發編程的高效，因此考慮將三個服務拆分，都分別由單獨的執行緒來執行
進一步把不同的服務部署到不同的服務器上，而且每個服務是通過服務器集群的形式來部署，從而形成了分布式的架構
- 這時不同服務之間的呼叫就需要引入rpc，因為是跨服務器跨網路的呼叫，不再是單機內部的方法呼叫

什么是RPC？

RPC（Remote Procedure Call ——遠程程序呼叫），它是一種通過網路從遠程計算機程式上請求服務，而不需要了解底層網路的技術，–來自《百度百科》

客戶端存根–server stub

扣減庫存服務，為了使呼叫遠端服務和呼叫本地服務沒有什么區別，就需要引入客戶端存根
客戶端存根就是遠程方法在本地的模擬，包括方法名、方法引數

服務端存根–client stub

負責把客戶端請求發來的請求體，決議成跟服務器實際方法相同的方法名、方法引數

總結

rpc方法就是把上述程序全部包裝起來，使得呼叫遠程方法就跟呼叫本地方法一樣

RPC和HTTP

TCP/IP概念層模型	功能	TCP/IP協議族
	檔案傳輸、電子郵件、檔案服務、虛擬終端	TFTP、HTTP、SNMP、FTP、SMTP、DNS、Telnet
應用層	資料格式化、代碼轉換、資料加密	無
	解除或建立與別的接點的聯系	無
傳輸層	提供端與端的介面	TCP、UDP
網路層	為資料包選擇路由	IP、ICMP、RIP、OSPF、BGP、IGMP
鏈路層	傳輸有地址的幀以及錯誤檢測功能	SLIP、CSLIP、PPP、ARP、RARP、MTU
	以二進制資料形式在物理媒體上傳輸資料	ISO2110、IEEE802、IEEE802.2

rpc究竟在上面協議分層的哪一層？

rpc只是一種思想，對不同服務呼叫的一種描述，既可以通過http實作，也可以通過tcp、htp，所以rpc和http不是一個層級的東西

實作RPC框架需要解決的那些問題？

通信問題
- 不要每次方法呼叫，都需要建立socket連接
代理問題
- 可以不可以通過代理的方式，每次的通信都通過指定的代理來解決
- 代理問題的解決
  
  代理模式，用動態代理，為什么不用靜態代理？-- 針對每一個服務都要創建相應的代理類，而rpc框架不知道要建立哪些靜態代理，所以使用動態代理，不管你要呼叫什么服務，通過動態代理一把解決
序列化問題
- 訊息在網路中傳輸是位元組的形式，怎么把01位元組變成javaBean的形式
- 序列化問題的解決
  
  實作Serializable介面，但是jdk的序列化性能很差
  
  測驗結果1-----jdk的位元組碼長度：133 自己位元組碼長度：24
  
  測驗結果2-----jdk序列化耗時：1364ms 自己序列化耗時：108ms
服務實體化
- 方法呼叫時只有一個方法名和方法引數，怎么轉化成呼叫具體的服務
- 服務實體化的解決
  
  通過反射解決

序列化耗時測驗

測驗1

package WangLuoBianCheng.wangLuoBianCheng3.rpc.prepare.serial.protogenesis;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;

/**
 * @author Mark老師   享學課堂 https://enjoy.ke.qq.com
 * 類說明：測驗序列化后位元組大小
 */
public class TestUserInfo {

    /**
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
		UserInfo info = new UserInfo();
		info.buildUserID(100).buildUserName("Welcome to Netty");
		//使用jdk的序列化
		ByteArrayOutputStream bos = new ByteArrayOutputStream();
		ObjectOutputStream os = new ObjectOutputStream(bos);
		os.writeObject(info);
		os.flush();
		os.close();
		byte[] b = bos.toByteArray();
		System.out.println("The jdk serializable length is : " + b.length);
		bos.close();
		//使用自行的序列化
		System.out.println("-------------------------------------");
		System.out.println("The byte array serializable length is : "
			+ info.codeC().length);
    }

}

測驗2

package WangLuoBianCheng.wangLuoBianCheng3.rpc.prepare.serial.protogenesis;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.nio.ByteBuffer;

/**
 * @author Mark老師   享學課堂 https://enjoy.ke.qq.com
 * 類說明：測驗序列化性能差異
 */
public class PerformTestUserInfo {

    public static void main(String[] args) throws IOException {
        UserInfo info = new UserInfo();
        info.buildUserID(100).buildUserName("Welcome to Netty");
        int loop = 1000000;

        //使用jdk的序列化
        ByteArrayOutputStream bos = null;
        ObjectOutputStream os = null;
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < loop; i++) {
            bos = new ByteArrayOutputStream();
            os = new ObjectOutputStream(bos);
            os.writeObject(info);
            os.flush();
            os.close();
            byte[] b = bos.toByteArray();
            bos.close();
        }
        long endTime = System.currentTimeMillis();
        System.out.println("The jdk serializable cost time is  : "
            + (endTime - startTime) + " ms");

        //使用自行的序列化
        System.out.println("-------------------------------------");
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        startTime = System.currentTimeMillis();
        for (int i = 0; i < loop; i++) {
            byte[] b = info.codeC(buffer);
        }
        endTime = System.currentTimeMillis();
        System.out.println("The byte array serializable cost time is : "
            + (endTime - startTime) + " ms");

        }

}

實作rpc框架

1.服務端定義介面和服務實作類并且注冊服務
2.客戶端使用動態代理呼叫服務(動態代理)
3.客戶端代理把呼叫物件、方法、引數序列化成資料
4.客戶端代理與服務端通過Socket通訊傳輸資料
5.服務端反序列化資料成物件、方法、引數，
6.服務端代理拿到這些物件和引數后通過反射的機制呼叫服務的實體，

將短信服務拆分成rpc服務—服務端

服務物體類，服務需要的相關引數型別 -----> 這些可以看出服務端存根

所有服務使用的服務框架----RpcServerFrame

package cn.enjoyedu.rpc.rpc.base;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;

/**
 *@author Mark老師  
 *
 *類說明：rpc框架的服務端部分
 */
@Service
public class RpcServerFrame {

//    @Autowired
//    private RegisterService registerService;
    @Autowired
    private RegisterServiceWithRegCenter registerServiceWithRegCenter;

    //服務的埠號
    private int port;

    /*處理服務請求任務*/
    private static class ServerTask implements Runnable{

        private Socket socket;
        private RegisterServiceWithRegCenter registerServiceWithRegCenter;

        public ServerTask(Socket client,
                          RegisterServiceWithRegCenter registerServiceWithRegCenter) {
            this.socket = client;
            this.registerServiceWithRegCenter = registerServiceWithRegCenter;
        }
		
        // 接受客戶端的請求，并呼叫實際的方法
        // 接受的內容---1.方法所在的類名介面名 2.呼叫的方法名  3.方法引數及方法屬性、具體的引數值
        // 
        @Override
        public void run() {
            try(
                 ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                  ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream())){

                /*方法所在類名介面名*/
                // 讀字串inputStream.readUTF()
                String serviceName = inputStream.readUTF();
                /*方法的名字*/
                String methodName = inputStream.readUTF();
                /*方法的入參型別*/
                // 讀物件inputStream.readObject()
                // 強制轉型，轉成類
                Class<?>[] paramTypes = (Class<?>[]) inputStream.readObject();
                /*方法的入參的值*/
                Object[] args = (Object[]) inputStream.readObject();

                /*從容器中拿到服務的Class物件*/
                Class serviceClass = registerServiceWithRegCenter.getLocalService(serviceName);
                if(serviceClass == null){
                    throw new ClassNotFoundException(serviceName+ " not found");
                }

                /*通過反射，執行實際的服務*/
                // 通過反射獲取具體的方法Method物件
                // 入參包括兩個，方法名和引數型別陣列
                Method method = serviceClass.getMethod(methodName, paramTypes);
                // 通過動態代理呼叫回應的方法
                Object result  = method.invoke(serviceClass.newInstance(),args);

                /*將服務的執行結果通知呼叫者*/
                // 將呼叫結果輸出給呼叫者，即客戶端
                outputStream.writeObject(result);
                outputStream.flush();

            }catch (Exception e){
                e.printStackTrace();
            }finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void startService(String serviceName, String host, int port, Class impl) throws Throwable{
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress(port));
        System.out.println("RPC server on:"+port+":運行");
       // registerService.regRemote(ServiceName, impl);
        registerServiceWithRegCenter.regRemote(serviceName,host,port,impl);
        try{
            while(true){
                new Thread(new ServerTask(serverSocket.accept(),
                        registerServiceWithRegCenter)).start();
            }
        }finally {
            serverSocket.close();
        }
    }

}

服務注冊中心

package cn.enjoyedu.rpc.rpc.base;

import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 類說明：注冊服務到本地快取
 */
@Service
public class RegisterService {

    /*本地可以提供服務的一個容器*/
    private static final Map<String,Class> serviceCache = new ConcurrentHashMap<>();

    /*注冊本服務*/
    public void regService(String serviceName,Class impl){
        serviceCache.put(serviceName,impl);
    }

    /*獲取服務*/
    public Class getLocalService(String serviceName){
        return serviceCache.get(serviceName);
    }


}

短信服務啟動

SmsRpcServer

package cn.enjoyedu.rpc.rpc.sms;

import cn.enjoyedu.rpc.rpc.base.RpcServerFrame;
import cn.enjoyedu.rpc.remote.SendSms;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Random;

/**
 * 類說明：rpc的服務端，提供服務
 */
@Service
public class SmsRpcServer {

    // rpc的base框架部分注入
    @Autowired
    private RpcServerFrame rpcServerFrame;

    @PostConstruct
    public void server() throws Throwable {
        Random r = new Random();
        // 埠號
        int port = 8778+r.nextInt(100);
        // 啟動服務
        rpcServerFrame.startService(SendSms.class.getName(),
                "127.0.0.1",port,SendSmsImpl.class);

    }

}

SendSmsImpl—具體的服務實作類

package cn.enjoyedu.rpc.rpc.sms;


import cn.enjoyedu.rpc.remote.vo.UserInfo;
import cn.enjoyedu.rpc.remote.SendSms;

/**
 *@author Mark老師   享學課堂 https://enjoy.ke.qq.com 
 *
 *類說明：短資訊發送服務的實作
 */
public class SendSmsImpl implements SendSms {

    @Override
    public boolean sendMail(UserInfo user) {
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("已發送短資訊給："+user.getName()+"到【"+user.getPhone()+"】");
        return true;
    }
}

將短信服務拆分成rpc服務—客戶端

類似于服務器端存根，也需要有客戶端存根

客戶端框架類

package cn.enjoyedu.rpc.client.rpc;

import cn.enjoyedu.rpc.remote.vo.RegisterServiceVo;
import org.springframework.stereotype.Service;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Set;

/**
 *@author Mark老師   
 *類說明：rpc框架的客戶端代理部分
 */
@Service
public class RpcClientFrame {

    /*遠程服務的代理物件，引數為客戶端要呼叫的的服務*/
    // 傳遞的應該是方法所在的類名或介面名
    public static<T> T getRemoteProxyObject(final Class<?> serviceInterface) throws Exception {
        /*獲得遠程服務的一個網路地址*/
        InetSocketAddress addr = //new InetSocketAddress("127.0.0.1",8778);
           getService(serviceInterface.getName());

        /*拿到一個代理物件，由這個代理物件通過網路進行實際的服務呼叫*/
        // 應該產生一個代理物件，和服務器進行通訊
        // 需要實作了InvocationHandler的類DynProxy
        return (T)Proxy.newProxyInstance(serviceInterface.getClassLoader(),
                new Class<?>[]{serviceInterface},
                new DynProxy(serviceInterface,addr));
    }


    /*動態代理，實作對遠程服務的訪問*/
    // 連接服務器，發送介面名、方法名、方法引數
    private static class DynProxy implements InvocationHandler{
        private Class<?> serviceInterface;
        private InetSocketAddress addr;

        public DynProxy(Class<?> serviceInterface, InetSocketAddress addr) {
            this.serviceInterface = serviceInterface;
            this.addr = addr;
        }

        @Override
        public Object invoke(Object proxy, Method method, Object[] args)
                throws Throwable {
            Socket socket = null;
            ObjectInputStream inputStream = null;
            ObjectOutputStream outputStream = null;
            try{
                
                // 1.建立網路連接
                socket = new Socket();
                socket.connect(addr);
                outputStream = new ObjectOutputStream(socket.getOutputStream());

                // 2.發送介面名、方法名、引數型別、引數值
                //方法所在類名介面名
                outputStream.writeUTF(serviceInterface.getName());
                //方法的名字
                outputStream.writeUTF(method.getName());
                //方法的入參型別
                outputStream.writeObject(method.getParameterTypes());
                //方法入參的值
                outputStream.writeObject(args);

                outputStream.flush();

                // 3.接受服務器傳回的結果
                inputStream = new ObjectInputStream(socket.getInputStream());
                /*接受服務器的輸出*/
                System.out.println(serviceInterface+" remote exec success!");
                return inputStream.readObject();

            }finally {
                if(socket!=null) socket.close();
                if(outputStream!=null) outputStream.close();
                if(inputStream!=null) inputStream.close();

            }
        }
    }



    /*----------------以下和動態獲得服務提供者有關------------------------------*/

    private static Random r = new Random();

    /*獲得遠程服務的地址*/
    private static InetSocketAddress getService(String serviceName)
            throws Exception {
        //獲得服務提供者的地址串列
        List<InetSocketAddress> serviceVoList = getServiceList(serviceName);
        InetSocketAddress addr
                = serviceVoList.get(r.nextInt(serviceVoList.size()));
        System.out.println("本次選擇了服務器："+addr);
        return addr;
    }

    /*獲得服務提供者的地址*/
    private static List<InetSocketAddress> getServiceList(String serviceName)
            throws Exception {
        Socket socket = null;
        ObjectOutputStream output = null;
        ObjectInputStream input = null;

        try{
            socket = new Socket();
            socket.connect(new InetSocketAddress("127.0.0.1",9999));

            output = new ObjectOutputStream(socket.getOutputStream());
            //需要獲得服務提供者
            output.writeBoolean(true);
            //告訴注冊中心服務名
            output.writeUTF(serviceName);
            output.flush();

            input = new ObjectInputStream(socket.getInputStream());
            Set<RegisterServiceVo> result
                    = (Set<RegisterServiceVo>)input.readObject();
            List<InetSocketAddress> services = new ArrayList<>();
            for(RegisterServiceVo serviceVo : result){
                String host = serviceVo.getHost();//獲得服務提供者的IP
                int port = serviceVo.getPort();//獲得服務提供者的埠號
                InetSocketAddress serviceAddr = new InetSocketAddress(host,port);
                services.add(serviceAddr);
            }
            System.out.println("獲得服務["+serviceName
                    +"]提供者的地址串列["+services+"]，準備呼叫.");
            return services;
        }finally{
            if (socket!=null) socket.close();
            if (output!=null) output.close();
            if (input!=null) input.close();
        }

    }

}

遠端的服務怎么通過bean接管

package cn.enjoyedu.rpc.client.config;

import cn.enjoyedu.rpc.client.rpc.RpcClientFrame;
import cn.enjoyedu.rpc.remote.SendSms;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 類說明：
 */
@Configuration
public class BeanConfig {

    @Autowired
    private RpcClientFrame rpcClientFrame;

    // 獲取服務端的方法名
    @Bean
    public SendSms getSmsService() throws Exception{
        return rpcClientFrame.getRemoteProxyObject(SendSms.class);
    }
}

rpc測驗

啟動服務端

package cn.enjoyedu.rpc;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RpcServerSmsApplication {

    public static void main(String[] args) {
        SpringApplication.run(RpcServerSmsApplication.class, args);
    }

}

客戶端測驗方法

package cn.enjoyedu.rpc.client;

import cn.enjoyedu.rpc.client.service.NormalBusi;
import cn.enjoyedu.rpc.remote.SendSms;
import cn.enjoyedu.rpc.remote.StockService;
import cn.enjoyedu.rpc.remote.vo.UserInfo;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class RpcClientApplicationTests {

    @Autowired
    private NormalBusi normalBusi;
//    @Autowired
//    private StockService stockService;
    @Autowired
    private SendSms sendSms;

    @Test
    void rpcTest() {
        long start = System.currentTimeMillis();
        normalBusi.business();

        /*發送郵件*/
        UserInfo userInfo = new UserInfo("Mark","Mark@qq.com");
        System.out.println("Send mail: "+ sendSms.sendMail(userInfo));
        System.out.println("共耗時："+(System.currentTimeMillis()-start)+"ms");


        /*扣減庫存*/
//        stockService.addStock("A001",1000);
//        stockService.deduceStock("B002",50);
    }

}

測驗總結

現在在客戶端只有SendSms這樣一個介面，并沒有實作類
把代理類作為實作類注入到spring的容器里面
代理類里面，把對方法的呼叫通過網路轉給了遠端短信發送服務器上
而遠端短信發送服務器上，通過一個map容器保存了改服務器上能提供的服務
在服務器端啟動rpc服務時，把具體的實作放到hashmap中，并且啟動了一個網路相關的服務，專門接受客戶端傳遞過來的方法呼叫請求
從而實作了一次rpc的遠程方法呼叫

注冊中心

dubbo等rpc框架存在一個注冊中心，而現在的rpc呼叫只有一個，如果要實作集群化，如果要提供多個rpc服務，難道客戶端呼叫的時候要寫多個服務器地址進去？
注冊中心本質也是提供rpc服務，只有兩種，提供服務的注冊，提供服務的查詢

RegisterCenter

package cn.enjoyedu.rpc.rpc.reg.service;

import cn.enjoyedu.rpc.remote.vo.RegisterServiceVo;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * @author Mark老師  
 * 類說明：服務注冊中心，服務提供者在啟動時需要在注冊中心登記自己的資訊
 */
@Service
public class RegisterCenter {
    /*key表示服務名，value代表服務提供者地址的集合*/
    private static final Map<String,Set<RegisterServiceVo>> serviceHolder
            = new HashMap<>();

    /*注冊服務的埠號*/
    private int port;

    /*服務注冊，考慮到可能有多個提供者同時注冊，進行加鎖*/
    private static synchronized void registerService(String serviceName,
                                String host,int port){
        //獲得當前服務的已有地址集合
        Set<RegisterServiceVo> serviceVoSet = serviceHolder.get(serviceName);
        if(serviceVoSet==null){
            //已有地址集合為空，新增集合
            serviceVoSet = new HashSet<>();
            serviceHolder.put(serviceName,serviceVoSet);
        }
        //將新的服務提供者加入集合
        serviceVoSet.add(new RegisterServiceVo(host,port));
        System.out.println("服務已注冊["+serviceName+"]，" +
                "地址["+host+"]，埠["+port+"]");
    }

    /*取出服務提供者*/
    private static Set<RegisterServiceVo> getService(String serviceName){
        return serviceHolder.get(serviceName);
    }

    /*處理服務請求的任務，其實無非就是兩種服務：
    1、服務注冊服務
    2、服務查詢服務
    */
    private static class ServerTask implements Runnable{
        private Socket client = null;

        public ServerTask(Socket client){
            this.client = client;
        }

        public void run() {

            try(ObjectInputStream inputStream =
                        new ObjectInputStream(client.getInputStream());
                ObjectOutputStream outputStream =
                        new ObjectOutputStream(client.getOutputStream())){

                /*檢查當前請求是注冊服務還是獲得服務*/
                boolean isGetService = inputStream.readBoolean();
                /*服務查詢服務，獲得服務提供者*/
                if(isGetService){
                    String serviceName = inputStream.readUTF();
                    /*取出服務提供者集合*/
                    Set<RegisterServiceVo> result = getService(serviceName);
                    /*回傳給客戶端*/
                    outputStream.writeObject(result);
                    outputStream.flush();
                    System.out.println("將已注冊的服務["+serviceName+"提供給客戶端");
                }
                /*服務注冊服務*/
                else{
                    /*取得新服務提供方的ip和埠*/
                    String serviceName = inputStream.readUTF();
                    String host = inputStream.readUTF();
                    int port = inputStream.readInt();
                    /*在注冊中心保存*/
                    registerService(serviceName,host,port);
                    outputStream.writeBoolean(true);
                    outputStream.flush();
                }
            }catch(Exception e){
                e.printStackTrace();
            }finally {
                try {
                    client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /*啟動注冊服務*/
    public void startService() throws IOException {
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress(port));
        System.out.println("服務注冊中心 on:"+port+":運行");
        try{
            while(true){
                new Thread(new ServerTask(serverSocket.accept())).start();
            }
        }finally {
            serverSocket.close();
        }
    }

    // 服務注冊中心埠是9999
    @PostConstruct
    public void init() {
        this.port = 9999;
        new Thread(new Runnable() {
            public void run() {
                try{
                    startService();
                }catch(IOException e){
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

測驗使用服務注冊中心

package cn.enjoyedu.rpc;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RpcRegApplication {

    public static void main(String[] args) {
        SpringApplication.run(RpcRegApplication.class, args);
    }

}

修改回應的服務提供者

之前的服務都是注冊在本地，現在除了在本地存一份，現在改成注冊到注冊中心，在哪個ip地址上提供了哪些服務

RegisterServiceWithRegCenter

package cn.enjoyedu.rpc.rpc.base;

import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 類說明：注冊服務,引入了服務的注冊和發現機制
 */
@Service
public class RegisterServiceWithRegCenter {

    /*本地可提供服務的一個名單，用快取實作*/
    private static final Map<String,Class> serviceCache
            = new ConcurrentHashMap<>();

    /*往遠程注冊服務器注冊本服務,同時在本地注冊本服務*/
    public void regRemote(String serviceName, String host, int port, Class impl)
            throws Throwable{
        //登記到注冊中心
        Socket socket = null;
        ObjectOutputStream output = null;
        ObjectInputStream input = null;

        try{
            socket = new Socket();
            socket.connect(new InetSocketAddress("127.0.0.1",9999));

            output = new ObjectOutputStream(socket.getOutputStream());
            /*注冊服務*/
            output.writeBoolean(false);
            /*提供的服務名*/
            output.writeUTF(serviceName);
            /*服務提供方的IP*/
            output.writeUTF(host);
            /*服務提供方的埠*/
            output.writeInt(port);
            output.flush();

            input = new ObjectInputStream(socket.getInputStream());
            if(input.readBoolean()){
                System.out.println("服務["+serviceName+"]注冊成功!");
            }

            /*可提供服務放入本地快取*/
            serviceCache.put(serviceName,impl);

        } catch (IOException e) {
            e.printStackTrace();
        }  finally{
            if (socket!=null) socket.close();
            if (output!=null) output.close();
            if (input!=null) input.close();
        }
    }

    /*獲取服務*/
    public Class getLocalService(String serviceName) {
        return serviceCache.get(serviceName);
    }

}

修改rpcServerFrame

package cn.enjoyedu.rpc.rpc.base;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;

/**
 *@author Mark老師  
 *
 *類說明：rpc框架的服務端部分
 */
@Service
public class RpcServerFrame {

//    @Autowired
//    private RegisterService registerService;
    // 修改1
    @Autowired
    private RegisterServiceWithRegCenter registerServiceWithRegCenter;

    //服務的埠號
    private int port;

    /*處理服務請求任務*/
    private static class ServerTask implements Runnable{

        private Socket socket;
        // 修改3
        private RegisterServiceWithRegCenter registerServiceWithRegCenter;

        // 修改4
        public ServerTask(Socket client,
                          RegisterServiceWithRegCenter registerServiceWithRegCenter) {
            this.socket = client;
            this.registerServiceWithRegCenter = registerServiceWithRegCenter;
        }
		
        // 接受客戶端的請求，并呼叫實際的方法
        // 接受的內容---1.方法所在的類名介面名 2.呼叫的方法名  3.方法引數及方法屬性、具體的引數值
        // 
        @Override
        public void run() {
            try(
                 ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                  ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream())){

                /*方法所在類名介面名*/
                // 讀字串inputStream.readUTF()
                String serviceName = inputStream.readUTF();
                /*方法的名字*/
                String methodName = inputStream.readUTF();
                /*方法的入參型別*/
                // 讀物件inputStream.readObject()
                // 強制轉型，轉成類
                Class<?>[] paramTypes = (Class<?>[]) inputStream.readObject();
                /*方法的入參的值*/
                Object[] args = (Object[]) inputStream.readObject();

                // 修改5
                /*從容器中拿到服務的Class物件*/
                Class serviceClass = registerServiceWithRegCenter.getLocalService(serviceName);
                if(serviceClass == null){
                    throw new ClassNotFoundException(serviceName+ " not found");
                }

                /*通過反射，執行實際的服務*/
                // 通過反射獲取具體的方法Method物件
                // 入參包括兩個，方法名和引數型別陣列
                Method method = serviceClass.getMethod(methodName, paramTypes);
                // 通過動態代理呼叫回應的方法
                Object result  = method.invoke(serviceClass.newInstance(),args);

                /*將服務的執行結果通知呼叫者*/
                // 將呼叫結果輸出給呼叫者，即客戶端
                outputStream.writeObject(result);
                outputStream.flush();

            }catch (Exception e){
                e.printStackTrace();
            }finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void startService(String serviceName, String host, int port, Class impl) throws Throwable{
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress(port));
        System.out.println("RPC server on:"+port+":運行");
        // 修改2
       // registerService.regRemote(ServiceName, impl);
	
              registerServiceWithRegCenter.regRemote(serviceName,host,port,impl);
        try{
            while(true){
                new Thread(new ServerTask(serverSocket.accept(),
                        registerServiceWithRegCenter)).start();
            }
        }finally {
            serverSocket.close();
        }
    }

}

測驗注冊中心

為了多注冊服務，埠號取隨機的
啟動sms
再啟動一次sms

客戶端啟動

動態代理的地址就不能寫死了，應該從注冊中心獲取地址

package cn.enjoyedu.rpc.client.rpc;

import cn.enjoyedu.rpc.remote.vo.RegisterServiceVo;
import org.springframework.stereotype.Service;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Set;

/**
 *@author Mark老師   
 *類說明：rpc框架的客戶端代理部分
 */
@Service
public class RpcClientFrame {

    /*遠程服務的代理物件，引數為客戶端要呼叫的的服務*/
    // 傳遞的應該是方法所在的類名或介面名
    public static<T> T getRemoteProxyObject(final Class<?> serviceInterface) throws Exception {
        /*獲得遠程服務的一個網路地址*/
        // 修改1
        // InetSocketAddress addr = new InetSocketAddress("127.0.0.1",8778);
        InetSocketAddress addr =   getService(serviceInterface.getName());

        /*拿到一個代理物件，由這個代理物件通過網路進行實際的服務呼叫*/
        // 應該產生一個代理物件，和服務器進行通訊
        // 需要實作了InvocationHandler的類DynProxy
        return (T)Proxy.newProxyInstance(serviceInterface.getClassLoader(),
                new Class<?>[]{serviceInterface},
                new DynProxy(serviceInterface,addr));
    }


    /*動態代理，實作對遠程服務的訪問*/
    // 連接服務器，發送介面名、方法名、方法引數
    private static class DynProxy implements InvocationHandler{
        private Class<?> serviceInterface;
        private InetSocketAddress addr;

        public DynProxy(Class<?> serviceInterface, InetSocketAddress addr) {
            this.serviceInterface = serviceInterface;
            this.addr = addr;
        }

        @Override
        public Object invoke(Object proxy, Method method, Object[] args)
                throws Throwable {
            Socket socket = null;
            ObjectInputStream inputStream = null;
            ObjectOutputStream outputStream = null;
            try{
                
                // 1.建立網路連接
                socket = new Socket();
                socket.connect(addr);
                outputStream = new ObjectOutputStream(socket.getOutputStream());

                // 2.發送介面名、方法名、引數型別、引數值
                //方法所在類名介面名
                outputStream.writeUTF(serviceInterface.getName());
                //方法的名字
                outputStream.writeUTF(method.getName());
                //方法的入參型別
                outputStream.writeObject(method.getParameterTypes());
                //方法入參的值
                outputStream.writeObject(args);

                outputStream.flush();

                // 3.接受服務器傳回的結果
                inputStream = new ObjectInputStream(socket.getInputStream());
                /*接受服務器的輸出*/
                System.out.println(serviceInterface+" remote exec success!");
                return inputStream.readObject();

            }finally {
                if(socket!=null) socket.close();
                if(outputStream!=null) outputStream.close();
                if(inputStream!=null) inputStream.close();

            }
        }
    }



    /*----------------以下和動態獲得服務提供者有關------------------------------*/

    private static Random r = new Random();

    /*獲得遠程服務的地址*/
    private static InetSocketAddress getService(String serviceName)
            throws Exception {
        //獲得服務提供者的地址串列
        List<InetSocketAddress> serviceVoList = getServiceList(serviceName);
        InetSocketAddress addr
                = serviceVoList.get(r.nextInt(serviceVoList.size()));
        System.out.println("本次選擇了服務器："+addr);
        return addr;
    }

    /*獲得服務提供者的地址*/
    private static List<InetSocketAddress> getServiceList(String serviceName)
            throws Exception {
        Socket socket = null;
        ObjectOutputStream output = null;
        ObjectInputStream input = null;

        try{
            socket = new Socket();
            socket.connect(new InetSocketAddress("127.0.0.1",9999));

            output = new ObjectOutputStream(socket.getOutputStream());
            //需要獲得服務提供者
            output.writeBoolean(true);
            //告訴注冊中心服務名
            output.writeUTF(serviceName);
            output.flush();

            input = new ObjectInputStream(socket.getInputStream());
            Set<RegisterServiceVo> result
                    = (Set<RegisterServiceVo>)input.readObject();
            List<InetSocketAddress> services = new ArrayList<>();
            for(RegisterServiceVo serviceVo : result){
                String host = serviceVo.getHost();//獲得服務提供者的IP
                int port = serviceVo.getPort();//獲得服務提供者的埠號
                InetSocketAddress serviceAddr = new InetSocketAddress(host,port);
                services.add(serviceAddr);
            }
            System.out.println("獲得服務["+serviceName
                    +"]提供者的地址串列["+services+"]，準備呼叫.");
            return services;
        }finally{
            if (socket!=null) socket.close();
            if (output!=null) output.close();
            if (input!=null) input.close();
        }

    }

}

取服務時，如果有多個，做負載均衡？

用亂數

高并發RPC解決方案

基于TCP的RPC實作
- Dubbo:
  阿里巴巴公司開源的一個高性能優秀的服務框架，使得應用可通過高性能的RPC實作服務的輸出和輸入功能，可以和Spring框架無縫集成，
Provider: 暴露服務的服務提供方，
Consumer: 呼叫遠程服務的服務消費方，
Registry: 服務注冊與發現的注冊中心，
Monitor: 統計服務的呼叫次調和呼叫時間的監控中心，
Container: 服務運行容器，

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/258699.html

標籤：其他

上一篇：Nginx+Tomcat實作負載均衡、動靜分離集群部署

下一篇：還在用Mybatis? Spring Data JPA 讓你的開發效率提升數倍！