Publisher/Subscriber(发布/订阅者)消息模式开发流程

实际上上 该模式的作用是发布者和订阅者 可以相互发送消息

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

演示代码

发布者

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) {
}

}
}

}

先运行 订阅者 在运行 发布者 试看效果