实现一个简单的软负载中心

package whu.edu.lcrpc.configserver.thread; /** * Created by apple on 17/4/20. */ import lombok.Data; import whu.edu.lcrpc.configserver.data.DataSingleClass; import whu.edu.lcrpc.configserver.entity.requestdata.*; import whu.edu.lcrpc.configserver.entity.responsedata.HeartBeatResponseDO; import whu.edu.lcrpc.configserver.entity.responsedata.ResponseDataDO; import whu.edu.lcrpc.configserver.entity.responsedata.ServiceIPsDO; import whu.edu.lcrpc.configserver.entity.responsedata.ServiceRegistryResultDO; import whu.edu.lcrpc.configserver.service.ISendData; import whu.edu.lcrpc.configserver.service.IUpdateData; import whu.edu.lcrpc.configserver.service.impl.SendDataImpl; import whu.edu.lcrpc.configserver.service.impl.UpdateDataImpl; import javax.annotation.Resource; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.net.Socket; import java.util.ArrayList; import java.util.UUID; /** * 处理socket请求连接线程 */ @Data public class DealWithConnectionThread implements Runnable{ private ISendData sendData = null; private IUpdateData updateData = null; private Socket socket = null;//长连接发送的数据包DO private DataSingleClass dataSingle = null;//软负载中心维护的三类数据的访问接口 public DealWithConnectionThread(Socket socket){ this.socket = socket; this.dataSingle = DataSingleClass.getData(); this.sendData = new SendDataImpl(); this.updateData = new UpdateDataImpl(); } @Override public void run() { //获取输入,序列化后,判断数据包类型,针对不同类型进行不同的处理 /** * 发送数据包的类型 * 0:服务注册数据包: 更新"聚合数据",并根据订阅关系,推送更新 * 1:服务发布端发送的心跳包: 更新该服务该地址的最后更新时间,即更新"聚合数据" * 2:服务调用端发送的心跳包 * 3:服务信息查询数据包: 更新"订阅关系",并推送服务地址列表 */ ObjectInputStream ois = null; ObjectOutputStream oos = null; boolean flag = true; try { ois = new ObjectInputStream(socket.getInputStream()); oos = new ObjectOutputStream(socket.getOutputStream()); while (flag){ ReqeustDataDO reqeustDataDO = (ReqeustDataDO) ois.readObject(); ResponseDataDO responseDataDO = null; //根据type值判断数据包的类型 switch (reqeustDataDO.type){ case 0: responseDataDO = serviceRegistry((ServiceRegistryDO) reqeustDataDO); break; case 1: responseDataDO = serviceProHeartbeat((ServiceProHeartbeatDO) reqeustDataDO); break; case 2: responseDataDO = new HeartBeatResponseDO(); break; case 3: responseDataDO = serviceInfoQuery((ServiceQueryDO) reqeustDataDO,socket,oos); break; default:break; } //发送响应数据 oos.writeObject(responseDataDO); } } catch (IOException e) { e.printStackTrace(); flag = false; } catch (ClassNotFoundException e) { e.printStackTrace(); flag = false; }finally { try { if (ois != null)ois.close(); if (oos != null)oos.close(); } catch (IOException e) { e.printStackTrace(); } } } /** * 收到服务注册请求,更新"聚合数据" * 若该服务存在 * @param serviceRegistryDO */ private ResponseDataDO serviceRegistry(ServiceRegistryDO serviceRegistryDO){ ServiceRegistryResultDO registryResultDO = new ServiceRegistryResultDO(); //step1.更新"聚合数据" String serviceID = serviceRegistryDO.getInterfaceName() + "_" + serviceRegistryDO.getVersion(); if (updateData.addServiceData(serviceRegistryDO.getInterfaceName(),serviceRegistryDO.getVersion(),serviceRegistryDO.getImplClassName(),serviceRegistryDO.getIp())){ //服务注册成功 System.out.println("[" + serviceID + "] 服务注册成功"); //step2. 判断该服务是否有订阅的客户端,若有,则发送该服务信息到所有订阅者 if (dataSingle.subscriptionData.containsKey(serviceID)){ sendData.sendServiceAddrToSubscriber(serviceID); } //step3. 增加连接 // String connID = socket.getInetAddress().getHostAddress(); // updateData.addConnection(connID,socket); // System.out.println("增加一个服务发布者连接[" + connID + "]"); }else { //服务注册失败 System.out.println("[" + serviceID + "] 服务注册失败"); registryResultDO.flag = false; } return registryResultDO; } private ResponseDataDO serviceProHeartbeat(ServiceProHeartbeatDO serviceProHeartbeatDO) { HeartBeatResponseDO heartBeatResponseDO = new HeartBeatResponseDO(); //更新该服务该地址的 心跳包接收时间 String serviceID = serviceProHeartbeatDO.getInterfaceName() + "_" + serviceProHeartbeatDO.getVersion(); // System.out.println("收到服务[" + serviceID + "]发来的心跳包"); heartBeatResponseDO.flag = (updateData.updateServiceAddrUpdateTime(serviceID,serviceProHeartbeatDO.getIp())); return heartBeatResponseDO; } private ResponseDataDO serviceInfoQuery(ServiceQueryDO serviceQueryDO, Socket socket,ObjectOutputStream oos){ ServiceIPsDO serviceIPsDO = new ServiceIPsDO(); serviceIPsDO.setServiceID(serviceQueryDO.getServiceID()); String serviceID = serviceQueryDO.getServiceID(); System.out.println("查询服务[" + serviceID + "]地址列表"); //step1. 查询该服务的地址列表 if (!dataSingle.serviceData.containsKey(serviceID)){ System.out.println("服务[" + serviceID + "不存在"); serviceIPsDO.flag = false; }else { //step2. 增加连接数据 String connID = socket.getInetAddress().getHostAddress() + "_" + UUID.randomUUID().toString(); updateData.addConnection(connID,socket,oos); System.out.println("增加一个服务调用者连接[" + connID + "]"); //step3. 更新订阅关系 updateData.addSubscription(serviceID,connID); System.out.println("服务[" + serviceID + "]增加一个订阅者[" + connID + "]"); //step4. 获取地址列表 serviceIPsDO.setIps(dataSingle.serviceData.get(serviceID).getIps().keySet()); } return serviceIPsDO; } }