activemq5.2发送和接收BlobMessage简单实例

package com.work.activemq;

/**
 * @author wangmingjie
 * @date 2009-7-29上午09:00:48
 */

import java.io.File;
import java.io.IOException;

import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.BlobMessage;
import org.apache.activemq.command.ActiveMQQueue;

public class BlobMessageSendTest {
  
    private String user = ActiveMQConnection.DEFAULT_USER;    
   
    private String password = ActiveMQConnection.DEFAULT_PASSWORD;    
   
//    private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private String url = "tcp://localhost:61616?jms.blobTransferPolicy.defaultUploadUrl=http://localhost:8161/fileserver/";
   
    private String subject = "Blob Queue";
   
    private Destination destination = null;    
   
    private ActiveMQConnection connection = null;    
   
    private ActiveMQSession session = null;    
   
    private MessageProducer producer = null;    
   
    // 初始化    
    private void initialize() throws JMSException, Exception {    
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(  url);    
        connection = (ActiveMQConnection)connectionFactory.createConnection();    
  /*
   * !!!!!!!!!!!!!!!!!!!!!!!!! very important. If it is set to true
   * (default) the uploader is lost in translation ;)
   * !!!!!!!!!!!!!!!!!!!!!!!!!
   */
        connection.setCopyMessageOnSend(false);        
        session = (ActiveMQSession)connection.createSession(false, Session.AUTO_ACKNOWLEDGE);    
        destination = session.createQueue(subject);
//        destination = session.createTopic(subject);
        producer = session.createProducer(destination);    
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);    
    }    
   
    // 发送消息    
    public void produceMessage(File file) throws JMSException, Exception {    
        initialize();    
        BlobMessage msg = session.createBlobMessage(file); 
        
        connection.start();    
        System.out.println("Producer:->Sending message: " + file.getName());    
        producer.send(msg);    
        System.out.println("Producer:->Message sent complete!");    
    }    
   
    // 关闭连接    
    public void close() throws JMSException {    
        System.out.println("Producer:->Closing connection");    
        if (producer != null)    
            producer.close();    
        if (session != null)    
            session.close();    
        if (connection != null)    
            connection.close();    
    }    
    
    
    
 /**
  * 测试成功，能够发送到服务器上。
  * @param args
  * @throws IOException
  */
 public void send() throws IOException {

  long start = System.currentTimeMillis();

  /*
   * First you must tell how the Blob repository can be found. Use
   * ActiveMQ 5.1.0+ as broker because the fileserver webapp works there
   * out of the box.
   */
  ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
    "tcp://localhost:61616?jms.blobTransferPolicy.defaultUploadUrl=http://localhost:8161/fileserver/");

  ActiveMQConnection conn = null;
  ActiveMQSession session = null;
  try {
   conn = (ActiveMQConnection) cf.createConnection();
   session = (ActiveMQSession) conn.createSession(false,
     Session.AUTO_ACKNOWLEDGE);
   Destination destination = new ActiveMQQueue("Blob Queue");
   MessageProducer producer = session.createProducer(destination);

   /*
    * !!!!!!!!!!!!!!!!!!!!!!!!! very important. If it is set to true
    * (default) the uploader is lost in translation ;)
    * !!!!!!!!!!!!!!!!!!!!!!!!!
    */
   conn.setCopyMessageOnSend(false);

//   File file = File.createTempFile("amq-data-file-", ".dat");
//   // lets write some data
//   BufferedWriter writer = new BufferedWriter(new FileWriter(file));
//   writer.append("Hello World!");
//   writer.close();
   File file = new File("d:/BUG管理需求和设计.doc");
   /*
    * i have used an simple ascii file instead of the picture you want
    * to use
    */
   BlobMessage message = session.createBlobMessage(file);

   /*
    * should only work if you receive the message
    */
   // System.out.println(message.getInputStream());
   producer.send(message);

   /*
    * After all you can see a new file under
    * $Brokerlocation$/webapps/fileserver/
    */
  } catch (JMSException e) {
   e.printStackTrace();
  } finally {
   try {
    if (session != null) {
     session.close();
    }
    if (conn != null) {
     conn.close();
    }
   } catch (JMSException ex) {
   }
  }
  System.out.println("共花费了"+(System.currentTimeMillis() - start)+"毫秒！");
 }

}

package com.work.activemq;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.BlobMessage;


/**
 * @author wangmingjie
 * @date 2009-7-29上午09:05:10
 */
public class BlobMessageReceiveTest  implements MessageListener {    
    private String user = ActiveMQConnection.DEFAULT_USER;    
    
    private String password = ActiveMQConnection.DEFAULT_PASSWORD;    
   
//  private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private String url = "tcp://localhost:61616?jms.blobTransferPolicy.defaultUploadUrl=http://localhost:8161/fileserver/";
   
    private String subject = "Blob Queue";    
   
    private Destination destination = null;    
   
    private ActiveMQConnection connection = null;    
   
    private ActiveMQSession session = null;    
   
    private MessageConsumer consumer = null;    
    
    private static int  count = 1; //计数器
   
    // 初始化    
    private void initialize() throws JMSException, Exception {    
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(    
               url);    
        connection = (ActiveMQConnection)connectionFactory.createConnection();
		/*
		 * !!!!!!!!!!!!!!!!!!!!!!!!! very important. If it is set to true
		 * (default) the uploader is lost in translation ;)
		 * !!!!!!!!!!!!!!!!!!!!!!!!!
		 */
        connection.setCopyMessageOnSend(false);          
        session = (ActiveMQSession)connection.createSession(false, Session.AUTO_ACKNOWLEDGE);    
        destination = session.createQueue(subject); 
//        destination = session.createTopic(subject);
        consumer = session.createConsumer(destination);    
            
    }    
   
    // 消费消息    
    public void consumeMessage() throws JMSException, Exception {    
        initialize();    
        connection.start();    
            
        System.out.println("Consumer:->Begin listening...");    
        // 开始监听    
        
        consumer.setMessageListener(this);    
        // Message message = consumer.receive();    
    }    
   
    // 关闭连接    
    public void close() throws JMSException {    
        System.out.println("Consumer:->Closing connection");    
        if (consumer != null)    
            consumer.close();    
        if (session != null)    
            session.close();    
        if (connection != null)    
            connection.close();    
    }    
   
    // 消息处理函数    
    public void onMessage(Message message) {    
        try {    
            if (message instanceof TextMessage) {    
                TextMessage txtMsg = (TextMessage) message;    
                String msg = txtMsg.getText();    
                System.out.println("Consumer:->Received: " + msg);    
            }else if (message instanceof BlobMessage) {
	    		try {			
					InputStream inStr = ((BlobMessage)message).getInputStream();
					BufferedInputStream bin = new BufferedInputStream(inStr);
					File file = new File("c:/new.rar");
					FileOutputStream fos = new FileOutputStream(file);
					
					byte[] block = new byte[4096];//大小不同，文件就可能损坏。
					while (true) {
						int readLength = bin.read(block);
						if (readLength == -1)
							break;// end of file
						fos.write(block, 0, readLength); //一定要写入实际读取内容
					}
					bin.close();
					inStr.close();
					fos.flush();
					fos.close();
					System.out.println("第"+count+++"次Consumer:->Received BlobMessage wangmj "); 
				} catch (IOException e) {
					e.printStackTrace();
				}	            	 
            }else  {   
                System.out.println("Consumer:->Received: " + message);    
            }    
        } catch (JMSException e) {    
            // TODO Auto-generated catch block    
            e.printStackTrace();    
        }    
    }    
    
}

package com.work.activemq;

import java.io.File;

/**
* @author wangmingjie
* @date 2009-7-29上午09:21:53
*/
public class BlobMessageClient {

/**
* topic方式，必须先启动消费者，然后是生产者，否则接收不到消息。
* queue方式，最好先启动生产者，然后启动消费者，否则也容易收不到消息。
* @param args
*/
public static void main(String[] args) throws Exception {

BlobMessageSendTest producer = new BlobMessageSendTest();
BlobMessageReceiveTest consumer = new BlobMessageReceiveTest();

String fileName = "e:/javagood/clearbug1.6.zip";
// String fileName = "d:/JAVA+开发视频会议系统详细设计.doc";
File file = new File(fileName);
producer.produceMessage(file);

producer.close();
// 延时500毫秒之后停止接受消息
Thread.sleep(2000);
// 开始监听
consumer.consumeMessage();
// 延时500毫秒之后发送消息
Thread.sleep(2000);
consumer.close();
}

}