JMS 使用 ActiveMQ 传送文件
JMS 使用 ActiveMQ 传送文件
(2010年04月11日) 发表于 Java博客
createBlobMessage(URL url)
createBlobMessage(URL url, boolean deletedByBroker)
createBlobMessage(File file)
createBlobMessage(InputStream in)
接收到 BlobMessage 消息后，可以调用其 getInputStream() 方法获得数据，然后写成磁盘文件，文件名、文件大小等可通过 Message 的 getXxxProperty("Property.Name") 取的。
注意，传输入文件的时候，发送方创建 ConnectionFactory 时的 brokerURL 需要指定 jms.blobTransferPolicy.uploadUrl 或者jms.blobTransferPolicy.defaultUploadUrl 属性为 ActiveMQ 中 fileserver 应用的 URI，即指定传输 BlogMessage 的 BlobTransferPolicy 策略，参看 Configuring the BLOB Transfer Policy。
先上图看真相，包括执行 FileSender 时选择文件，FileReciever 在收到文件时会弹出保存文件的对话框，还有控制台的打印信息。
具体来看例子，使用 P2P 队列的方式，你也可以用发布/订阅的方式，只是代码中创建的 Destination 类型不同。
1. 启动 ActiveMQ
在 http://activemq.apache.org/activemq-531-release.html 下载 ActiveMQ，然后解压，执行其中的 bin/activemq.bat，控制台可以看到：
INFO | ActiveMQ Console at http://0.0.0.0:8161/admin
INFO | Initializing Spring root WebApplicationContext
INFO | Successfully connected to tcp://localhost:61616
INFO | Camel Console at http://0.0.0.0:8161/camel
INFO | ActiveMQ Web Demos at http://0.0.0.0:8161/demo
INFO | RESTful file access application at http://0.0.0.0:8161/fileserver
INFO | Started [email protected]:8161
表示启动成功了，假设运行 ActiveMQ 机器的 IP 地址是 10.80.38.10。
2. 编写发送文件的程序 FileSender.java
package com.unmi.jms;
import java.io.File;
import javax.jms.*;
import javax.swing.JFileChooser;
import org.apache.activemq.*;
/**
* 通过 ActiveMQ 发送文件的程序
* @author Unmi
*/
public class FileSender {
/**
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
// 选择文件
JFileChooser fileChooser = new JFileChooser();
fileChooser.setDialogTitle("请选择要传送的文件");
if (fileChooser.showOpenDialog(null) != JFileChooser.APPROVE_OPTION) {
return;
}
File file = fileChooser.getSelectedFile();
// 获取 ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
"tcp://10.80.38.10:61616?jms.blobTransferPolicy.defaultUploadUrl=http://10.80.38.10:8161/fileserver/");
// 创建 Connection
Connection connection = connectionFactory.createConnection();
connection.start();
// 创建 Session
ActiveMQSession session = (ActiveMQSession) connection.createSession(
false, Session.AUTO_ACKNOWLEDGE);
// 创建 Destination
Destination destination = session.createQueue("File.Transport");
// 创建 Producer
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 设置为非持久性
// 设置持久性的话，文件也可以先缓存下来，接收端离线再连接也可以收到文件
// 构造 BlobMessage，用来传输文件
BlobMessage blobMessage = session.createBlobMessage(file);
blobMessage.setStringProperty("FILE.NAME", file.getName());
blobMessage.setLongProperty("FILE.SIZE", file.length());
System.out.println("开始发送文件：" + file.getName() + "，文件大小："
+ file.length() + " 字节");
// 7. 发送文件
producer.send(blobMessage);
System.out.println("完成文件发送：" + file.getName());
producer.close();
session.close();
connection.close(); // 不关闭 Connection, 程序则不退出
}
}
3. 编写接收文件的程序 FileReceiver.java
package com.unmi.jms;
import java.io.*;
import javax.jms.*;
import javax.jms.Message;
import javax.swing.*;
import org.apache.activemq.*;
public class FileReciever {
/**
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
// 获取 ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
"tcp://10.80.38.10:61616");
// 创建 Connection
Connection connection = connectionFactory.createConnection();
connection.start();
// 创建 Session
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
// 创建 Destinatione
Destination destination = session.createQueue("File.Transport");
// 创建 Consumer
MessageConsumer consumer = session.createConsumer(destination);
// 注册消息监听器，当消息到达时被触发并处理消息
consumer.setMessageListener(new MessageListener() {
// 监听器中处理消息
public void onMessage(Message message) {
if (message instanceof BlobMessage) {
BlobMessage blobMessage = (BlobMessage) message;
try {
String fileName = blobMessage.getStringProperty("FILE.NAME");
System.out.println("文件接收请求处理：" + fileName + "，文件大小："
+ blobMessage.getLongProperty("FILE.SIZE")+ " 字节");
JFileChooser fileChooser = new JFileChooser();
fileChooser.setDialogTitle("请指定文件保存位置");
fileChooser.setSelectedFile(new File(fileName));
if (fileChooser.showSaveDialog(null) == JFileChooser.APPROVE_OPTION) {
File file = fileChooser.getSelectedFile();
OutputStream os = new FileOutputStream(file);
System.out.println("开始接收文件：" + fileName);
InputStream inputStream = blobMessage.getInputStream();
//写文件，你也可以使用其他方式
byte[] buff = new byte[256];
int len = 0;
while ((len = inputStream.read(buff)) > 0) {
os.write(buff, 0, len);
}
os.close();
System.out.println("完成文件接收：" + fileName);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
});
}
}
4. 运行程序
先执行 FileReceiver 程序来监听消息
再执行发送程序 FileSender，将会提示你选择一个要传送的文件(上图左边)，确定后就会把文件发送到 ActiveMQ 服务器上
接收端 FileReceiver 监听到有文件传过来的消息后，会自动弹出保存文件的对话框，要你选择保存位置(上图右边)，文件名能保持一致。确定后就开始接收文件，存到指定的位置。
如果设置 producer.setDeliveryMode(DeliveryMode.PERSISTENT); 消息持久性的话，发送方传文件的时候，接收方可以不在线，文件会暂存在 ActiveMQ 服务器上，等到接收程序上线后仍然可以收到发过来的文件。
参考：1. A problem with blobmessage on activemq 5.0
2. Apache ActiveMQ > Features > Message Features > Blob Messages
3. Apache ActiveMQ > Features > Message Dispatching Features > Async Sends
[版权声明]本站内文章，如未特别注明，均系原创或翻译之作，本人 Unmi 保留一切权利。本站原创及译作未经本人许可，不得用于商业用途及传统媒体。网络媒体可随意转载，或以此为基础进行演译，但务必以链接形式注明原始出处和作者信息，否则属于侵权行为。另对本站转载他处文章，俱有说明，如有侵权请联系本人，本人将会在第一时间删除侵权文章。及此说明，重之之重。