FastDFS连接池实现_fastdfsconnectionpool.getstorageserver().getinetso

连接池实现

package com.ccqtgb;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;


import org.csource.fastdfs.ClientGlobal;
import org.csource.fastdfs.StorageClient;
import org.csource.fastdfs.StorageClient1;
import org.csource.fastdfs.StorageServer;
import org.csource.fastdfs.TrackerClient;
import org.csource.fastdfs.TrackerGroup;
import org.csource.fastdfs.TrackerServer;


public class ConnectionPool {


    //最大连接数,可以写配置文件
    private int size = 5;
    //被使用的连接
    private ConcurrentHashMap<StorageClient1,Object> busyConnectionPool = null;
    //空闲的连接
    private ArrayBlockingQueue<StorageClient1> idleConnectionPool = null;
    
    private Object obj = new Object();
    
    private static ConnectionPool instance = new ConnectionPool();
    
    public static ConnectionPool getConnectionPool(){
        return instance;
    }
    
    //取出连接
    public StorageClient1 checkout(int waitTime){
        StorageClient1 storageClient1 = null;
        try {
            storageClient1 = idleConnectionPool.poll(waitTime, TimeUnit.SECONDS);
System.out.println(storageClient1);            
            if(storageClient1 != null){
                busyConnectionPool.put(storageClient1, obj);
            }
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            storageClient1 = null;
            e.printStackTrace();
        }
        return storageClient1;
    }
    
    //回收连接
    public void checkin(StorageClient1 storageClient1){
        if(busyConnectionPool.remove(storageClient1) != null){
            idleConnectionPool.add(storageClient1);
        }
    }
    
    //如果连接无效则抛弃，新建连接来补充到池里
    public void drop(StorageClient1 storageClient1){
        if(busyConnectionPool.remove(storageClient1) != null){
            TrackerServer trackerServer = null;
            TrackerClient trackerClient = new TrackerClient();
            try {
                trackerServer = trackerClient.getConnection();
                StorageClient1 newStorageClient1 = new StorageClient1(trackerServer,null);
                idleConnectionPool.add(newStorageClient1);
System.out.println("------------------------- :connection +1");
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }finally{
                if(trackerServer != null){
                    try {
                        trackerServer.close();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        }
    }
    
    //单例
    private ConnectionPool(){
        busyConnectionPool = new ConcurrentHashMap<StorageClient1, Object>();
        idleConnectionPool = new ArrayBlockingQueue<StorageClient1>(size);
        init(size);
    }
    
    //初始化连接池
    private void init(int size){
        initClientGlobal();
        TrackerServer trackerServer = null;
        try {
            TrackerClient trackerClient = new TrackerClient();
            //只需要一个tracker server连接
            trackerServer = trackerClient.getConnection();
            StorageServer storageServer = null;
            StorageClient1 storageClient1 = null;
            for(int i=0; i<size; i++){
                storageClient1 = new StorageClient1(trackerServer,storageServer);
                idleConnectionPool.add(storageClient1);
System.out.println("------------------------- :connection +1");
            }
            
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally{
            if(trackerServer != null){
                try {
                    trackerServer.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
    }
    
    //初始化客户端
    private void initClientGlobal(){
        //连接超时时间
        ClientGlobal.setG_connect_timeout(2000);
        //网络超时时间
        ClientGlobal.setG_network_timeout(3000);
        ClientGlobal.setG_anti_steal_token(false);
        // 字符集
        ClientGlobal.setG_charset("UTF-8");
        ClientGlobal.setG_secret_key(null);
        // HTTP访问服务的端口号
        ClientGlobal.setG_tracker_http_port(8080);
         
        InetSocketAddress[] trackerServers = new InetSocketAddress[2];
        trackerServers[0] = new InetSocketAddress("10.64.2.171",22122);
        trackerServers[1] = new InetSocketAddress("10.64.2.172",22122);
        TrackerGroup trackerGroup = new TrackerGroup(trackerServers);
        //tracker server 集群
        ClientGlobal.setG_tracker_group(trackerGroup);
    }
    
    

}

============================================

客户端接口

package com.ccqtgb.client;


import java.io.File;


public interface FileClient {


    public String uploadFile(File file);
    
    public String uploadFile(File file, String name);
    
    public String uploadFile(byte[] buff, String name);
    
}

=====================

客户端实现

package com.ccqtgb.client.impl;


import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;


import org.csource.common.NameValuePair;
import org.csource.fastdfs.StorageClient1;


import com.ccqtgb.ConnectionPool;
import com.ccqtgb.client.FileClient;


public class FileClientImpl implements FileClient {


    @Override
    public String uploadFile(File file) {
        byte[] buff = getFileBuff(file);
        String file_ext_name =  getFileExtName(file.getName());
        return send(buff, file_ext_name, null);
    }


    @Override
    public String uploadFile(File file, String file_ext_name) {
        byte[] buff = getFileBuff(file);
        return send(buff,file_ext_name,null);
    }


    @Override
    public String uploadFile(byte[] buff,String file_ext_name) {
        return send(buff,file_ext_name,null);
    }
    
    //上传缓存数据到storage服务器
    private String send(byte[] buff, String file_ext_name, NameValuePair[] mate_list){
        String upPath = null;
        StorageClient1 storageClient1 = null;
        ConnectionPool pool = ConnectionPool.getConnectionPool();
        storageClient1 = pool.checkout(10);
        try {
            upPath = storageClient1.upload_file1(buff, file_ext_name, mate_list);
            pool.checkin(storageClient1);
        } catch (IOException e) {
            //如果出现了IO异常应该销毁此连接
            pool.drop(storageClient1);
            e.printStackTrace();
        } catch (Exception e) {
            pool.drop(storageClient1);
            e.printStackTrace();
        }
        
        return upPath;
    }
    
    //将文件缓存到字节数组中
    private byte[] getFileBuff(File file){
        byte[] buff = null;
        try {
            FileInputStream inputStream = new FileInputStream(file);
            buff = new byte[inputStream.available()];
            inputStream.read(buff);
        } catch (FileNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return buff;
    }
    
    //通过文件名称获取文件扩展名
    private String getFileExtName(String name){
        String ext_name = null;
        if(name != null){
            if(name.contains(".")){
                ext_name = name.substring(name.indexOf(".")+1);
            }
        }
        return ext_name;
    }


}