使用ActiveMQ传输文件时出现的问题

    我在使用ActiveMQ传输文件时，采用的是如下方案：
     首先，对文件按字节读取，每次读取一定的字节数比如4096个字节放入字节数组中，然后，将该字节数组封装成字节型的消息发送，发送完成后再读取下一个字节段，一直到读取完毕为止。由于系统考虑到在传输大文件时，可能会由于网络中断等原因造成的文件传输中断，在文件的传输时实现了断点续传的功能，当每次发送成功一条消息后将当前读取的字节位置保存到一个临时文件中，当网络中断时，程序中止，下次启动该程序时先读取临时文件，跳过已读的字节数，再接着读取发送。网络中断可能发生在程序运行的任何地方，本程序中用到了ActiveMQ提供的事务管理，当消息发送以后事务提交并写临时文件，出现异常时事务回滚，在测试的过程中发现，如果是在事务提交语句 session.commit(); 执行时网络中断，此时的事务回滚代码执行了，但是消息好像仍然发送出去了，不知道这是我程序的问题，还是ActiveMQ的问题，在她的官方网站上正好这部分的文档打不开（提示域名不存在），这一个星期都在整这一个问题，真是郁闷了，有哪位仁兄对ActiveMQ有研究的话，给小弟指点一二，弟当感激不尽！

	private void sendFile(File file) throws Exception {
		long currentRead = this.ReadTempFile(file.getPath());  // 冲临时文件中读取当前传输的字节数
		long start = System.currentTimeMillis();
		RandomAccessFile rand = new RandomAccessFile(file,"r");
		rand.seek(currentRead);		// 跳过已读的字节
		byte[] buffer = new byte[2048];
		int c = -1;
		BytesMessage message = null;
		while ((c = rand.read(buffer)) >= 0) {
				try {
					message = session.createBytesMessage();
					message.setStringProperty("fileName", file.getPath()); //
					message.setLongProperty("fileLength", file.length());
					message.writeBytes(buffer, 0, c);
					System.out.println("Send: " + c);

					producer.send(message);	// 发送消息
					System.out.println("- 1 -");
					this.WriteTempFile(file.getPath(), c);	// 写临时文件
					System.out.println("- 2 -");

					if (transacted) {
						session.commit(); // 此处抛出异常时，运行结果与预期不符
						System.out.println("- 3 -");
					}
				} catch (JMSException e1) {
					session.rollback();
					System.err.println("- 4 -");
				}

			try {
				Thread.sleep(30);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		rand.close();
//		this.deleteTempFile(file.getPath());	// 删除临时文件
		long end = System.currentTimeMillis();
		System.out.println("Use time: " + (end - start));
	}

评论

2 楼 wxq276 2009-11-02  

可以呀，我试过了，代码如下：
发送端：

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

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.StreamMessage;

import org.apache.activemq.ActiveMQConnectionFactory;


public class StreamMsgProducer {
	public void send(File file){
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection conn = null;
		try{
			conn = factory.createConnection();
			conn.start();
			Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination queue = session.createQueue("streamMsg");
			MessageProducer producer = session.createProducer(queue);
			
			InputStream in = new FileInputStream(file);
			byte[] buffer = new byte[2048];
			int c = -1;
			while((c = in.read(buffer)) > 0){
				StreamMessage smsg = session.createStreamMessage();
				smsg.writeBytes(buffer, 0, c);
				producer.send(smsg);
				System.out.println("send: "  + c);
			}
			in.close();
		}catch(JMSException e){
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			if(conn != null){
				try {
					conn.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	public static void main(String[] args) {
		File file = new File("c:\\send.txt");
		new StreamMsgProducer().send(file);
	}
}

接收端：

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.StreamMessage;

import org.apache.activemq.ActiveMQConnectionFactory;


public class StreamMsgConsumer {
	public void receive(){
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection conn = null;
		try{
			conn = factory.createConnection();
			conn.start();
			Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination queue = session.createQueue("streamMsg");
			MessageConsumer consumer = session.createConsumer(queue);
			
			OutputStream out = new FileOutputStream("c:\\receive.txt");
			byte[] buffer = new byte[2048];
			while(true){
				Message msg = consumer.receive(5000);
				if(msg == null){
					break;
				}
				
				if(msg instanceof StreamMessage){
					StreamMessage smsg = (StreamMessage) msg;
					int c = smsg.readBytes(buffer);
					out.write(buffer, 0, c);
					System.out.println("Receive: " + c);
				}
			}
			out.close();
		}catch(JMSException e){
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			if(conn != null){
				try {
					conn.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	public static void main(String[] args) {
		new StreamMsgConsumer().receive();
	}
}

1 楼 江东孤鹤 2009-11-02  

请教一个问题，为什么我用StreamMessage时，就不能用buffer读文件呢，而BytesMessage可以，谢谢