ActiveMQ 中ActiveMQBlobMessage的接收和发送

           在ActiveMQ中对比较大的消息采用一种ActiveMQBlobMessage方式发送的时候，因为可采用策略的不同而已使用HTTP协议字节流，文件系统的读文件，FTP协议的方式实现发送和接收文件。

上传和下载的数据流执行过程：

           在ActiveMQ中采用策略模式定义如下接口：

上传策略接口BlobUploadStrategy：

public interface BlobUploadStrategy {

    URL uploadFile(ActiveMQBlobMessage message, File file) throws JMSException, IOException;

    URL uploadStream(ActiveMQBlobMessage message, InputStream in) throws JMSException, IOException;
}

 

下载策略接口BlobDownloadStrategy 如下：

public interface BlobDownloadStrategy {
    
    InputStream getInputStream(ActiveMQBlobMessage message) throws IOException, JMSException;
    
    void deleteFile(ActiveMQBlobMessage message) throws IOException, JMSException;

}

 

通过文件方式实现上传和下载的策略为FileSystemBlobStrategy ：

其中FileSystemBlobStrategy implements实现了 BlobUploadStrategy, BlobDownloadStrategy通过将源代码如下：

    protected File getFile(ActiveMQBlobMessage message) throws JMSException, IOException {
    	if (message.getURL() != null) {
    		try {
				return new File(message.getURL().toURI());
			} catch (URISyntaxException e) {
                                IOException ioe = new IOException("Unable to open file for message " + message);
                                ioe.initCause(e);
			}
    	}
        //replace all : with _ to make windows more happy
        String fileName = message.getJMSMessageID().replaceAll(":", "_"); 
        return new File(rootFile, fileName);        
        
    }

 

实现FTP上传和下载的策略的是FTPBlobUploadStrategy，FTPBlobDownloadStrategy实现。实质是采用commons-net包中的FTPClient实现相关的功能。

接口实现如下：

public class FTPBlobDownloadStrategy extends FTPStrategy implements BlobDownloadStrategy 

 

public class FTPBlobUploadStrategy extends FTPStrategy implements BlobUploadStrategy 

 

上传和下载均继承自FTPStrategy的实现如下：

package org.apache.activemq.blob;

import java.io.IOException;
import java.net.ConnectException;
import java.net.MalformedURLException;
import java.net.URL;

import javax.jms.JMSException;

import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.commons.net.ftp.FTPClient;

public class FTPStrategy {

    protected BlobTransferPolicy transferPolicy;
    protected URL url;
    protected String ftpUser = "";
    protected String ftpPass = "";

    public FTPStrategy(BlobTransferPolicy transferPolicy) throws MalformedURLException {
        this.transferPolicy = transferPolicy;
        this.url = new URL(this.transferPolicy.getUploadUrl());
    }
    
    protected void setUserInformation(String userInfo) {
        if(userInfo != null) {
            String[] userPass = userInfo.split(":");
            if(userPass.length > 0) this.ftpUser = userPass[0];
            if(userPass.length > 1) this.ftpPass = userPass[1];
        } else {
            this.ftpUser = "anonymous";
            this.ftpPass = "anonymous";
        }
    }
    
    protected FTPClient createFTP() throws IOException, JMSException {
        String connectUrl = url.getHost();
        setUserInformation(url.getUserInfo());
        int port = url.getPort() < 1 ? 21 : url.getPort();
        
        FTPClient ftp = new FTPClient();
        try {
            ftp.connect(connectUrl, port);
        } catch(ConnectException e) {
            throw new JMSException("Problem connecting the FTP-server");
        }
        if(!ftp.login(ftpUser, ftpPass)) {
            ftp.quit();
            ftp.disconnect();
            throw new JMSException("Cant Authentificate to FTP-Server");
        }
        return ftp;
    }
    
}

 

基于HTTP协议的默认

public class DefaultBlobUploadStrategy extends DefaultStrategy implements BlobUploadStrategy 

 

public class DefaultBlobDownloadStrategy extends DefaultStrategy implements BlobDownloadStrategy 

 

 

部分源代码如下：

public URL uploadFile(ActiveMQBlobMessage message, File file) throws JMSException, IOException {
        return uploadStream(message, new FileInputStream(file));
    }

    public URL uploadStream(ActiveMQBlobMessage message, InputStream fis) throws JMSException, IOException {
        URL url = createMessageURL(message);

        HttpURLConnection connection = (HttpURLConnection)url.openConnection();
        connection.setRequestMethod("PUT");
        connection.setDoOutput(true);

        // use chunked mode or otherwise URLConnection loads everything into
        // memory
        // (chunked mode not supported before JRE 1.5)
        connection.setChunkedStreamingMode(transferPolicy.getBufferSize());

        OutputStream os = connection.getOutputStream();

        byte[] buf = new byte[transferPolicy.getBufferSize()];
        for (int c = fis.read(buf); c != -1; c = fis.read(buf)) {
            os.write(buf, 0, c);
            os.flush();
        }
        os.close();
        fis.close();

        if (!isSuccessfulCode(connection.getResponseCode())) {
            throw new IOException("PUT was not successful: " + connection.getResponseCode() + " "
                                  + connection.getResponseMessage());
        }

        return url;
    }