189 8069 5689
websphere mq : 用于传输信息 具有跨平台的功能。
创新互联是一家集网站建设,潮南企业网站建设,潮南品牌网站建设,网站定制,潮南网站建设报价,网络营销,网络优化,潮南网站推广为一体的创新建站企业，帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿，时刻以成就客户成长自我，坚持不断学习、思考、沉淀、净化自己，让我们为更多的企业打造出实用型网站。
1 安装websphere mq 并启动
2 websphere mq 建立 queue Manager (如：MQSI_SAMPLE_QM)
3 建立queue 类型选择 Local类型 的 （如lq ）
3 建立channels 类型选择Server Connection （如BridgeChannel）
java 代码如下：
package test.mq;
import com.ibm.mq.*;
/*
* 成功的访问mq 的java 类
*/
public class FirstMqTest {
// public static void main(String[] args[]){
// FirstMqTest first = new FirstMqTest();
// first.test();
// }
public static void main(String args[]){
FirstMqTest first = new FirstMqTest();
first.test();
}
public void test(){
String qManager = "MQSI_SAMPLE_QM"; //QueueManager name
String qName = "lq";//Queue Name
try {
//configure connection parameters
MQEnvironment.hostname="172.16.17.123";//MQ Server name or IP
//MQEnvironment.port=1414;//listenr port
MQEnvironment.channel="BridgeChannel";//Server-Connection Channel
MQEnvironment.CCSID =1381;
// Create a connection to the QueueManager
System.out.println("Connecting to queue manager: "+qManager);
MQQueueManager qMgr = new MQQueueManager(qManager);
// Set up the options on the queue we wish to open
int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;
// Now specify the queue that we wish to open and the open options
System.out.println("Accessing queue: "+qName);
MQQueue queue = qMgr.accessQueue(qName, openOptions);
// Define a simple WebSphere MQ Message ...
MQMessage msg = new MQMessage();
// ... and write some text in UTF8 format
msg.writeUTF("Hello, World!");
// Specify the default put message options
MQPutMessageOptions pmo = new MQPutMessageOptions();
// Put the message to the queue
System.out.println("Sending a message...");
/*
* 在此测试一下 mq 的传输次列
*
*/
for(int j=0;j 5;j++){
String str ="test11111111111";
str = str+j;
msg.writeUTF(str);
queue.put(msg, pmo);
}
queue.put(msg, pmo);
// Now get the message back again. First define a WebSphere MQ message
// to receive the data
MQMessage rcvMessage = new MQMessage();
// Specify default get message options
MQGetMessageOptions gmo = new MQGetMessageOptions();
// Get the message off the queue.
System.out.println("...and getting the message back again");
queue.get(rcvMessage, gmo);
// And display the message text...
String msgText = rcvMessage.readUTF();
System.out.println("The message is: " + msgText);
// Close the queue
System.out.println("Closing the queue");
queue.close();
// Disconnect from the QueueManager
System.out.println("Disconnecting from the Queue Manager");
qMgr.disconnect();
System.out.println("Done!");
}
catch (MQException ex) {
System.out.println("A WebSphere MQ Error occured : Completion Code "
+ ex点抗 pletionCode + " Reason Code " + ex.reasonCode);
}
catch (java.io.IOException ex) {
System.out.println("An IOException occured whilst writing to the message buffer: "
+ ex);
}
}
}
ActiveMQ持久化消息的二种方式；
1、持久化为文件
这个装ActiveMQ时默认就是这种，只要设置消息为持久化就可以了。涉及到的配置和代码有：
persistenceAdapter
kahaDB directory="${activemq.base}/data/kahadb"/
/persistenceAdapter
producer.Send(request, MsgDeliveryMode.Persistent, level, TimeSpan.MinValue);
2、持久化为MySql
首先需要把MySql的驱动放到ActiveMQ的Lib目录下，我用的文件名字是：mysql-connector-java-5.0.4-bin.jar
接下来修改配置文件
persistenceAdapter
jdbcPersistenceAdapter dataDirectory="${activemq.base}/data" dataSource="#derby-ds"/
/persistenceAdapter
在配置文件中的broker节点外增加
bean id="derby-ds" class="org.apache点抗 mons.dbcp.BasicDataSource" destroy-method="close"
property name="driverClassName" value="com.mysql.jdbc.Driver"/
property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/
property name="username" value="activemq"/
property name="password" value="activemq"/
property name="maxActive" value="200"/
property name="poolPreparedStatements" value="true"/
/bean
从配置中可以看出数据库的名称是activemq，需要手动在MySql中增加这个库。
然后重新启动消息队列，会发现多了3张表
1：activemq_acks
2：activemq_lock
3：activemq_msgs
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import com.fxun.plant.vo.CommandVO;
public class ProducerTool extends Thread {
private Destination destination;
// private int messageCount = 500;
long sleepTime = 0;
// private boolean verbose = true;
// private int messageSize = 255;
private long timeToLive = 0; // 消息存活时间
private String user = ActiveMQConnection.DEFAULT_USER;
private String password = ActiveMQConnection.DEFAULT_PASSWORD;
private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private String subject; // subject的名字,默认是TOOL.DEFAULT
// private boolean topic;
private boolean transacted = false; // 是否采用事务
// private boolean persistent = false;
private P2PQueue p2pQueue;
public ProducerTool(String user, String password, String url, String subject) {
this.user = user;
this.password = password;
this.url = url;
this.subject = subject;
}
public void run() {
Connection connection = null;
try {
// Create the connection.
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
connection = connectionFactory.createConnection();
connection.start();
// Create the session
Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue(subject);
// Create the producer.
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
producer.setTimeToLive(timeToLive);
CommandVO commandVO = null;
int size = 0;
while (true) {
size = p2pQueue.getSize();
if (size 0) {
BytesMessage message = session.createBytesMessage();
message.writeInt(size);
for (int i = 0; i size; i++) {
commandVO = p2pQueue.pool();
if(commandVO == null) {
message.writeInt(0);
} else {
message.writeInt(commandVO.getCountSize());
message.writeInt(commandVO.getCommand()); // 指令
message.writeBytes(commandVO.getContent()); // 内容
}
}
producer.send(message);
}
Thread.sleep(300);
}
// Use the ActiveMQConnection interface to dump the connection
// stats.
// ActiveMQConnection c = (ActiveMQConnection) connection;
// c.getConnectionStats().dump(new IndentPrinter());
} catch (Exception e) {
System.out.println("Caught: " + e);
e.printStackTrace();
} finally {
try {
connection.close();
} catch (Throwable ignore) {
}
}
}
public void setPassword(String pwd) {
this.password = pwd;
}
public void setSleepTime(long sleepTime) {
this.sleepTime = sleepTime;
}
public void setSubject(String subject) {
this.subject = subject;
}
public void setTimeToLive(long timeToLive) {
this.timeToLive = timeToLive;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public void setUrl(String url) {
this.url = url;
}
public void setUser(String user) {
this.user = user;
}
public P2PQueue getP2pQueue() {
return p2pQueue;
}
public void setP2pQueue(P2PQueue p2pQueue) {
this.p2pQueue = p2pQueue;
}
}
原代码都发给你