1.下载ActiveMQ:
http://activemq.apache.org/download-archives.html
2.运行ActiveMQ
解压缩apache-activemq-5.5.1-bin.zip,然后双击apache-activemq-5.5.1\bin\win64\activemq.bat运行ActiveMQ程序。
注意:期间若遇到无法启动的情况,请尝试将计算机名改为全英文,不能有其他字符!
启动以后可浏览器打开一下网址登录,以便后面查看后面程序运行以后的信息!包括消息队列,消息出列,消费者等等信息!
http://localhost:8161/admin/ (默认用户名和密码是:admin admin)
3.编写producer和consumer。
ActiveMQ成功部署运行后,就可以开发producer和consumer了。
在eclipse中新建Java项目,将ActiveMQ解压目录下的jar包都引入项目(apache-activemq-5.13.0\activemq-all-5.13.0.jar),然后就可以编写producer和consumer。
生产者代码如下:
package isa.qa.activemqtest; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class Sender { public static void main(String[] args) { //ConnectionFactory :连接工厂,JMS用它创建连接 ConnectionFactory connectionFactory; //Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; //Session: 一个发送或接收消息的线程 Session session; // Destination :消息的目的地;消息发送给谁. Destination destination; // MessageProducer:消息发送者 MessageProducer producer; // TextMessage message; // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616" ); try { //构造从工厂得到连接对象 connection = connectionFactory.createConnection(); //启动 connection.start(); //获取操作连接 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值是一个服务器的queue/topic // destination = session.createQueue("FirstQueue"); destination = session.createTopic("FirstTopic"); // 得到消息生成者【发送者】 producer = session.createProducer(destination); // 持久化的设置,此处学习,实际根据项目决定 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 构造消息,此处写死,项目就是参数,或者方法获取 sendMessage(session, producer); session.commit(); } catch (Exception e) { e.printStackTrace(); } finally { try { if(null != connection){ connection.close(); } } catch (Throwable ignore) { } } } public static void sendMessage(Session session, MessageProducer producer) throws Exception{ TextMessage message = session.createTextMessage("ActiveMq发送的消息"); //发送消息到目的地方 System.out.println("ActiveMq发送消息"); producer.send(message); } }
消费者代码如下:
package isa.qa.activemqtest; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class Receiver { public static void main(String[] args) { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory; // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; // Session: 一个发送或接收消息的线程 Session session; // Destination :消息的目的地;消息发送给谁. Destination destination; // 消费者,消息接收者 MessageConsumer consumer; connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616"); try { // 构造从工厂得到连接对象 connection = connectionFactory.createConnection(); // 启动 connection.start(); // 获取操作连接 session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值是一个服务器的queue/topic // destination = session.createQueue("FirstQueue"); destination = session.createTopic("FirstTopic"); consumer = session.createConsumer(destination); while (true) { //设置接收者接收消息的超时时间 TextMessage message = (TextMessage) consumer.receive(500000); if (null != message) { System.out.println("收到消息" + message.getText()); } else { break; } } } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) { } } } }
4.注意:
生产者与消费者之间点对点只需设置:
destination = session.createQueue("FirstQueue");
生产者与消费者之间一对多只需设置:
destination = session.createTopic("FirstTopic");
时间: 2024-10-08 20:41:23