发布者和订阅者都充当 生产者和消费者

发布者

package publisher.to.subscriber;

import java.awt.font.TextMeasurer;

import javax.jms.Connection;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import org.apache.activemq.command.ActiveMQTopic;

public class Publisher implements MessageListener {

/**

* @param args

* @throws JMSException

*/

public static void main(String[] args) throws JMSException {

// TODO Auto-generated method stub

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();

Connection connection = factory.createConnection();

connection.start();

// 第一个参数设置是否需要事务支持

Session session = connection.createSession(true,

Session.AUTO_ACKNOWLEDGE);

// 发送消息 而 订阅者接受消息必须有个topic 名字也叫 msg.send

Destination sendTopic = new ActiveMQTopic("msg.message");

// 接受消息 而 订阅者发送消息必须有个topic 名字也叫 msg.receive

Destination sendReceive = new ActiveMQTopic("msg.control");

MessageProducer producer = session.createProducer(sendTopic);

MessageConsumer consumer = session.createConsumer(sendReceive);

Rec rec = new Rec(consumer,session,connection,producer);

rec.start();

}

public void onMessage(Message msg) {

// TODO Auto-generated method stub

}

}

class Rec extends Thread {

private MessageConsumer consumer = null;

private Session session = null;

private Connection connection=null;

MessageProducer producer=null;

public Rec(MessageConsumer consumer, Session session,Connection connection,MessageProducer producer) {

this.consumer = consumer;

this.session = session;

this.connection=connection;

this.producer=producer;

}

@Override

public void run() {

// TODO Auto-generated method stub

while (true) {

try {

TextMessage smsg = session.createTextMessage("hello my name is liaomin");

producer.send(smsg);

session.commit();

TextMessage msg = (TextMessage) consumer.receive();

System.out.println(msg.getText());

// 接收时必须提交以下 否则会出现上次收到的旧数据

session.commit();

session.close();

connection.close();

break;

} catch (Exception e) {

}

}

}

}

订阅者

package publisher.to.subscriber;

import javax.jms.Connection;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import org.apache.activemq.command.ActiveMQTopic;

public class Subscriber {

/**

* @param args

* @throws JMSException

*/

public static void main(String[] args) throws JMSException {

// TODO Auto-generated method stub

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();

Connection connection = factory.createConnection();

connection.start();

// 第一个参数设置是否需要事务支持

Session session = connection.createSession(true,

Session.AUTO_ACKNOWLEDGE);

//订阅者接受消息

Destination receiveTopic = new ActiveMQTopic("msg.message");

//订阅者发送消息

Destination sendTopic = new ActiveMQTopic("msg.control");

MessageProducer producer = session.createProducer(sendTopic);

MessageConsumer consumer = session.createConsumer(receiveTopic);

Rec1 rec = new Rec1(consumer,session,connection,producer);

rec.start();

}

public void onMessage(Message msg) {

// TODO Auto-generated method stub

}

}

class Rec1 extends Thread {

private MessageConsumer consumer = null;

private Session session = null;

private Connection connection;

private MessageProducer producer;

public Rec1(MessageConsumer consumer, Session session,Connection connection,MessageProducer producer) {

this.consumer = consumer;

this.session = session;

this.connection=connection;

this.producer=producer;

}

@Override

public void run() {

// TODO Auto-generated method stub

while (true) {

try {

TextMessage msg = (TextMessage) consumer.receive();

session.commit();

System.out.println(msg.getText());

TextMessage remsg = session.createTextMessage("ok receive");

producer.send(remsg);

// 接收时必须提交以下 否则会出现上次收到的旧数据

session.commit();

session.close();

connection.close();

break;

} catch (Exception e) {

}

}

}

}

先运行 订阅者 在运行 发布者