mqjava代码 java中的mq

求Java连接MQ的代码，通过代码实现收发消息，谢谢各位大神！

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;

}

}

原代码都发给你

java 怎么样调用IBM MQ 或者通信问题

websphere mq : 用于传输信息 具有跨平台mqjava代码的功能。

1 安装websphere mq 并启动

2 websphere mq 建立 queue Manager (如mqjava代码：MQSI_SAMPLE_QM)

3 建立queue 类型选择 Local类型 的 （如lq ）

3 建立channels 类型选择Server Connection （如BridgeChannel）

java 代码如下mqjava代码：

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.completionCode + " Reason Code " + ex.reasonCode);

}

catch (java.io.IOException ex) {

System.out.println("An IOException occured whilst writing to the message buffer: "

+ ex);

}

}

}

用java代码如何设置activemq消息持久化到数据库中？

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.commons.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

关于mqjava代码和java中的mq的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。