首页 > 代码库 > 个人第一个开源分布式项目distributeTemplate的实现三网络通讯netty传输大文件

个人第一个开源分布式项目distributeTemplate的实现三网络通讯netty传输大文件

2024-07-15 04:35:38 219人阅读

今天我将讲讲网络通讯，这里我初始版本由于采用的事Netty框架所以这里讲网络Netty在我们这里是怎么使用的，下周开始添加rpc lucene内容了实现之后的0.2 0.3版本，后面将会去掉netty依赖采用原生的NIO2 （aio）异步非阻塞方式实现自己网络通讯，也就是说这部分可能会实现一个简单的但是比netty精简高效的网络框架，后期做出来可能会单独开一个分支开源出来，netty说白了就是事件驱动以及 NIO 加一些协议以及异常处理，废话不多说了。

我最近有个想法就是想添加一个分布式锁的功能我最近正在试图去实现分布式锁，因为最近要做RPC 顺便去试图实现以下分布式锁试试，当然初级的排它锁先实现掉，后面的全程队列也就是rabbitmq那些消息中间件那样的话也会去试图实现的

前2天我加了一个配置类，为了更好的实现用户对配置的管理

public  class ConfigBuilder {

	private static ConfigBuilder configBuilder=new ConfigBuilder();
	//主机列表
	private List<InetSocketAddress> address=Lists.newArrayList();
	
	private InetSocketAddress currentHost=null;
	
	private List<ServiceConfig> services=Lists.newArrayList();
	//别名
	protected BiMap<String,InetSocketAddress> alias=HashBiMap.create();
	
	private String configFile;
	
    public ConfigBuilder() {
    	configFile=Context.defaultConfig;
    	Config config=ConfigFactory.load(Context.defaultConfig);
    	initConfig(config);
    }

	private void initConfig(Config config) {
		List<? extends ConfigObject> nodes=config.getObjectList("server.hosts");
    	for (ConfigObject node : nodes) {
    		  Integer remotePort=Integer.parseInt(node.get("remotePort").render());
   		      String remoteIp=node.get("remoteHost").unwrapped().toString();//远程主机的ip
   		      String name=node.containsKey("name")?node.get("name").unwrapped().toString():remoteIp;//主机别名
   		      InetSocketAddress host=new InetSocketAddress(remoteIp, remotePort);
   		      address.add(host);
    	      alias.put(name, host);
    	      //TODO 获取
    	      for(ServiceType serviceType : ServiceConfig.ServiceType.values()){
    	    	   String serviceName=serviceType.name().toLowerCase();
    	    	  if(node.containsKey(serviceName)){
    	    		  HashMap fcs=(HashMap) node.get(serviceName).unwrapped();
    	    		  ServiceConfig serviceConfig=new ServiceConfig();
    	    		  serviceConfig.setServiceType(serviceType);
    	    		  serviceConfig.setInfo(fcs);
    	    		  services.add(serviceConfig);
    	    	  }
    	      }
    	     
    	}
    	String chost=config.getString("client.currentHost");
		int port=config.getInt("client.currentPort");
		currentHost=new InetSocketAddress(chost, port);
	}
	
	public ConfigBuilder(String configFile) {
		this.configFile=configFile;
		Config config=ConfigFactory.load(configFile);
		initConfig(config);
	}
	public static ConfigBuilder getInstance(){
		if(configBuilder==null){
			return new ConfigBuilder();
		}else{
			return configBuilder;
		}
		
	}
	public static ConfigBuilder getInstance(String configFile){
		if(configBuilder==null){
			return new ConfigBuilder(configFile);
		}else{
			return configBuilder;
		}
		
	}
	/**
	 * 添加服务器主机
	 * @param host
	 * @param port
	 * @return
	 * 添加（修改）人：zhuyuping
	 */
	public ConfigBuilder addHost(String host,int port){
		InetSocketAddress h=new InetSocketAddress(host, port);
		address.add(h);
		alias.put(host, h);
		return this;
	}
	/**
	 * 添加主机并设置 主机别名
	 * @param host
	 * @param port
	 * @param nname
	 * @return
	 * 添加（修改）人：zhuyuping
	 */
    public ConfigBuilder addHost(String host,int port,String nname){
	      addHost(host, port);
	      alias.put(nname, new InetSocketAddress(host, port));
		return this;
	}
    /**
     * 移除主机
     * @param host
     * @param port
     * @return
     * 添加（修改）人：zhuyuping
     */
    public ConfigBuilder removeHost(String host,int port){
    	InetSocketAddress h=new InetSocketAddress(host, port);
		alias.inverse().remove(h);
		address.remove(h);
		return this;
	}
    /**
     * 设置当前自身的主机名 与端口 
     * @param host
     * @param port
     * @return
     * 添加（修改）人：zhuyuping
     */
    public ConfigBuilder setSelf(String host,int port){
		currentHost=new InetSocketAddress(host, port);
		return this;
	}
    /**
     * 写入到配置文件中
     * 
     * 添加（修改）人：zhuyuping
     */
    public void buildToFile(){
    	
    	String path=Thread.currentThread().getContextClassLoader().getResource("")+configFile;
    	try {
			Writer out = new BufferedWriter(new OutputStreamWriter(
					new FileOutputStream(path), "UTF8"));
			StringBuilder sb=new StringBuilder("server{");
			sb.append("\n\t").append("hosts=[");
			int i=0;
			int size=address.size();
			for (InetSocketAddress host : address) {
				sb.append("\t {");
				sb.append("name=").append(alias.inverse().get(host)).append("\n\t");
				sb.append("remoteHost=").append(host.getHostString()).append("\n\t");
				sb.append("remotePort=").append(host.getPort()).append("\n\t");
				 sb.append("\t }");
				    if(i!=size){
				    	sb.append(",");
				    }
			}
			sb.append("]").append("\n\t").append("}");	
			  //继续 保存client
			  
			   sb.append("\n\t").append("client{").append("\n\t").append("currentHost=").append(currentHost.getHostString()).append("\n\t");
			   sb.append("\n\t").append("currentPort=").append(currentHost.getPort()).append("\n\t");
			   sb.append("}");
			   out.write(sb.toString());
		} catch (Exception e) {
		
		}
    	
    	
    }
    
    public IConfig bulid(Context context){
    	
    	
    	return null;
    }
	
    
	
	@Override
	public String toString() {
		return "ConfigBuilder [address=" + address + ", currentHost="
				+ currentHost + ", services=" + services + ", alias=" + alias
				+ ", configFile=" + configFile + "]";
	}

	public static void main(String[] args) {
		ConfigBuilder build=ConfigBuilder.getInstance();
		build.addHost("192.168.8.8", 1234).addHost("192.168.9.9", 2222, "test");
		System.out.println(build);
	}
}

这样后面用户就很容易的建立配置文件了，本周末我会添加这样的方式到 distributeTemplate上，这样既可以用query对象也可以传递sql 进行相应的增删改查与同步。

ok 我们看看我们网络通讯怎么实现的

首先配置上下文都好，我们想想网络通讯是不是核心贯穿到处用到，既然这样我们何不做成 service 服务呢

所以我定义了一个服务

 
/**
 * 
 *      
 *     
 * @author zhuyuping       
 * @version 1.0     
 * @created 2014-7-9 下午12:56:04 
 * @function:把 一些 常用的网络通讯作为基础 通过其他客服端 驱动 来进行基本访问的 功能  封装成服务 交由 服务来解析 
 */
public interface Service {
    /**
     * 启动服务
     * 
     * 添加（修改）人：zhuyuping
     */
void start();
/**
 * 关闭服务 
 * 
 * 添加（修改）人：zhuyuping
 */
void stop(); 
/**
 * 服务的名称
 * @return
 * 添加（修改）人：zhuyuping
 */
String getName();
void setName(String name);
/**
 * 获得配置 
 * @return
 * 添加（修改）人：zhuyuping
 */
IConfig getConfig();
/**
 * 注入配置
 */
void setConfig(IConfig config);
    /**
     * 往主机中 新增文件
     * @param address
     * @param build
     * 添加（修改）人：zhuyuping
     */
void insert(List<InetSocketAddress> address,
ImmutableMap<String, Object> build);
    /**
     * 删除文件 或文件夹
     * @param address
     * @param build
     * 添加（修改）人：zhuyuping
     */
void delete(List<InetSocketAddress> address,
ImmutableMap<String, Object> build);
    /**
     * 同步文件或者文件夹 如果用户没有传递条件 那么默认为同步区域
     * @param address
     * @param build
     * 添加（修改）人：zhuyuping
     */
void sync(List<InetSocketAddress> address,
ImmutableMap<String, Object> build);
}

然后这个服务交给我们要用的时候注入进去就可以，这样就是我要网络那么网络服务来了，我要女朋友然后女朋友没来我要 xxx服务服务就能获取，而这个服务只在初始框架时候初始化进去，这样的情况，我们于是想到了一种模式

想到了吧就是

Flyweight 模式

就是这样，因为服务跟节点依赖了，所以我就节点这样采用了，后面我还要想想要不要脱离

public IConfig getConfig() {
return config;
}
public void setConfig(IConfig config) {
this.config = config;
}
    
public NodeFactory(IConfig config) {
super();
this.config = config;
}
private IConfig config;
private static Map<String,Node> caches=Maps.newHashMap();
public  Node createNode(Class<? extends Node> clazz){
Node nd=null;
if(clazz.isAssignableFrom(FileSystem.class)){
nd=caches.get(FileSystem.NAME);
if(nd==null){
nd=new FileSystem(config);
caches.put(FileSystem.NAME, nd);
}
}//....
return nd;
}
}

也就是说用Map 做缓存避免创建过多对象浪费内存，这里这样的情况很多场合都用到了，特别是连接池

下面进入正题我们看看怎么实现通讯，事件驱动监听到时间后，在队列轮训然后交给服务处理

然后这部分交给服务管理服务只提供访问发送数据而已

所以服务里面就需要一些访问通讯需要的如果你采用Tcp BIO 就只需要简单的创建 socket 以及serverSocket就可以了下面就用它来发送对象文件了了，关于netty 怎么创建基本使用可以参照我的博文 spring 与 netty简单整合，哪里有一个问题就是eventLoop没有分离开来知道的朋友，记得分开来，我就懒得重更新了，

我们来看服务里方法怎么发送的

这里检查是否连接上，然后发送没连接发送到死信队列里面，由于死信队列是弱引用然后在交给虚幻引用，最后回到道虚幻引用的队列里面，最后持久化记录下来，这类似于mysql的日志方式，用户如果若引用能够遍历到就去若引用找，不能遍历到就去虚幻引用的队列里面持久化回收的找，备份还原同步都用到他了，这里还有个就是网络连接为什么我不建立从新连接机制呢，其实netty建立重新连接很容易只需要在连接时候加个监听重新连接就可以，但是现在还不需要，大家理解就好，

关于tcp udp 服务端客服端见我代码里面server client包文件

小文件发送我们自定义一个协议编码就好，

 * 
 *      
 *     
 * @author zhuyuping       
 * @version 1.0     
 * @created 2014-7-15 下午7:00:49 
 * @function:将文件转为字节码传输
 */
public class FileEncoder extends MessageToByteEncoder<FileDataMessage>{
@Override
protected void encode(ChannelHandlerContext ctx, FileDataMessage msg,
ByteBuf out) throws Exception {
//协议 
//---head开始
// 1.head 2位 short int  2
//1.sesionid 4位 int     4
//2.hash  4为 int        4
//3.pathlenth 为 多少位？ 4
//4.blenth 为多少位 ?     4
//8. start 4字节                4
//9. end 4字节                   4
//----body开始
//5.command 1个字节          1
//6. path byte[]        ？
//7. byte[]             ?
   
msg.toBuffer(out);
//out.writeBytes(msg.toBuffer());
}
}

小文件解码

这里发送文件我采用的是netty官方提供的那个实例的2个文件，但是我修改了下，

大文件就不行了因为udp tcp 字节数的限制，虽然有个FixLenthXXX....可以拼接但是他依然不能上传大于2G的文件，

看看我是怎么实现的我覆盖了netty的源码类然后实现了了流上传的方法

/**
 * 
 *      
 *     
 * @author zhuyuping       
 * @version 1.0     
 * @created 2014-7-28 
 * @function：覆盖掉netty的源文件，实现大文件分割后的 上传
 */
public class MyHttpPostRequestEncoder implements ChunkedInput<HttpContent>{

	 /**
     * Different modes to use to encode form data.
     */
    public enum EncoderMode {
        /**
         *  Legacy mode which should work for most. It is known to not work with OAUTH. For OAUTH use
         *  {@link EncoderMode#RFC3986}. The W3C form recommentations this for submitting post form data.
         */
        RFC1738,

        /**
         * Mode which is more new and is used for OAUTH
         */
        RFC3986
    }

    private static final Map<Pattern, String> percentEncodings = new HashMap<Pattern, String>();

    static {
        percentEncodings.put(Pattern.compile("\\*"), "%2A");
        percentEncodings.put(Pattern.compile("\\+"), "%20");
        percentEncodings.put(Pattern.compile("%7E"), "~");
    }

    /**
     * Factory used to create InterfaceHttpData
     */
    private final HttpDataFactory factory;

    /**
     * Request to encode
     */
    private final HttpRequest request;

    /**
     * Default charset to use
     */
    private final Charset charset;

    /**