重庆分公司,新征程启航
为企业提供网站建设、域名注册、服务器等服务
ActiveMQ官方下载地址:http://activemq.apache.org/download.html
创新互联主营定西网站建设的网络公司,主营网站建设方案,重庆APP开发,定西h5小程序开发搭建,定西网站营销推广欢迎定西等地区企业咨询
下载解压后,执行bin目录下activemq.bat启动mq服务
找到lib目录并引入相关jar包
Sender消息发送者类
package com.ycj.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageProducer; import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; public class Sender { private static final String MQ_USER = "admin"; private static final String MQ_PWD = "admin"; private static final String MQ_URL = "tcp://localhost:61616"; private static final String MQ_QUEUE_DEFAULT = "testQueue"; public static void main(String[] args) throws JMSException { // 创建JMS连接工厂 ConnectionFactory factory = new ActiveMQConnectionFactory(MQ_USER, MQ_PWD, MQ_URL); // 通过工厂获取链接并启动 Connection connection = factory.createConnection(); connection.start(); // 创建session Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建destination 消息目的地 Destination destination = session.createQueue(MQ_QUEUE_DEFAULT); // 根据destination初始化一个生产者 MessageProducer producer = session.createProducer(destination); // 生产者消息不持久化 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 生成消息 并发送 Message msg1 = session.createTextMessage("I'm msg1"); Message msg2 = session.createTextMessage("I'm msg2"); producer.send(msg1); producer.send(msg2); session.commit(); connection.close(); System.out.println("producer send msg done!"); } }
Receiver消息接收者类
package com.ycj.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class Receiver { private static final String MQ_USER = "admin"; private static final String MQ_PWD = "admin"; private static final String MQ_URL = "tcp://localhost:61616"; private static final String MQ_QUEUE_DEFAULT = "testQueue"; public static void main(String[] args) throws JMSException { // 创建JMS连接工厂 ConnectionFactory factory = new ActiveMQConnectionFactory(MQ_USER, MQ_PWD, MQ_URL); // 通过工厂获取链接并启动 Connection connection = factory.createConnection(); connection.start(); // 创建session Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建destination 消息目的地 Destination destination = session.createQueue(MQ_QUEUE_DEFAULT); // 根据destination初始化一个消费者 MessageConsumer consumer = session.createConsumer(destination); while (true) { TextMessage msg = (TextMessage) consumer.receive(1000*1000); if (null != msg) { System.out.println("message has been received:" + msg.getText()); } } } }
依次运行Receiver和Sender,控制台输出:
Sender
Receiver
打开浏览器输入http://localhost:8161/admin/index.jsp,选择Queues可以监控队列消费情况