JMS-activeMq发布订阅模式(非持久订阅)

Publisher的代码:

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Publisher {
// 单例模式
// 1、连接工厂
private ConnectionFactory connectionFactory;
// 2、连接对象
private Connection connection;
// 3、Session对象
private Session session;
// 4、生产者
private MessageProducer messageProducer;
public Publisher() {
try {
this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
"123", "tcp://localhost:61616");
this.connection = connectionFactory.createConnection();
this.connection.start();
// 不使用事务
// 设置客户端签收模式
this.session = this.connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
this.messageProducer = this.session.createProducer(null);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public Session getSession() {
return this.session;
}
public void send1(/* String QueueName, Message message */) {
try {
Destination destination = this.session.createTopic("topic1");
MapMessage msg1 = this.session.createMapMessage();
msg1.setString("name", "张三");
msg1.setInt("age", 22);
MapMessage msg2 = this.session.createMapMessage();
msg2.setString("name", "李四");
msg2.setInt("age", 25);
MapMessage msg3 = this.session.createMapMessage();
msg3.setString("name", "张三");
msg3.setInt("age", 30);
// 发送消息到topic1
this.messageProducer.send(destination, msg1,
DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
this.messageProducer.send(destination, msg2,
DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
this.messageProducer.send(destination, msg3,
DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public void send2() {
try {
Destination destination = this.session.createTopic("topic1");
TextMessage message = this.session.createTextMessage("我是一个字符串");
// 发送消息
this.messageProducer.send(destination, message,
DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public static void main(String[] args) {
Publisher producer = new Publisher();
producer.send1();
}
}

Subscribe的代码:

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Subscriber {
// 单例模式
// 1、连接工厂
private ConnectionFactory connectionFactory;
// 2、连接对象
private Connection connection;
// 3、Session对象
private Session session;
// 4、生产者
private MessageConsumer messageConsumer;
// 5、目的地址
private Destination destination;
public Subscriber() {
try {
this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
"123", "tcp://localhost:61616");
this.connection = connectionFactory.createConnection();
this.connection.start();
// 不使用事务
// 设置客户端签收模式
this.session = this.connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
this.destination = this.session.createTopic("topic1");
this.messageConsumer = this.session.createConsumer(destination);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public Session getSession() {
return this.session;
}
// 用于监听消息队列的消息
class MyLister implements MessageListener {
@Override
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
}
if (message instanceof MapMessage) {
MapMessage ret = (MapMessage) message;
System.out.println(ret.toString());
System.out.println(ret.getString("name"));
System.out.println(ret.getInt("age"));
// 因为设置的是客户端的签收模式，所以要手动的去确认消息的消费
message.acknowledge();
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
// 用于异步监听消息
public void receiver() {
try {
this.messageConsumer.setMessageListener(new MyLister());
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public static void main(String[] args) {
Subscriber conmuser = new Subscriber();
conmuser.receiver();
}
}

先启动消费者(先订阅后消费)，再启动发布者

巴特西

JMS-activeMq发布订阅模式(非持久订阅)

最新文章

热门文章