来源:坏嘻嘻 发布时间:2018-09-18 15:36:44 阅读量:1166
介绍 Active MQ是个消息队列管理器,用于通讯的中间件。现在由我来给大家介绍一下如何使用ActiveMQ测试小程序。
修改bin目录下activemq.bat
在最后一行加入
1 | set ACTIVEMQ_OPTS=-Xms1G -Xmx1G |
保存后执行该bat文件
新建maven项目
pom文件中加入依赖
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | <dependencies> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <!--我的mq版本为5.9.0--> <version>5.9.0</version> </dependency> <!-- https: //mvnrepository.com/artifact/org.apache.activemq/activemq-broker --> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-broker</artifactId> <version>5.10.0</version> </dependency> <!-- https: //mvnrepository.com/artifact/org.apache.activemq/activemq-client --> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-client</artifactId> <version>5.14.0</version> </dependency> </dependencies> |
发送端程序
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 | import org.apache.activemq.ActiveMQConnection;import org.apache.activemq. ActiveMQConnectionFactory;import javax.jms.*;public class Send { // private static final int SEND_NUMBER = 10; 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注意参数值test是一个服务器的queue,须在ActiveMQ的console配置 destination = session.createQueue( "test" ); //得到发送者 producer = session.createProducer(destination); //设置不持久化,实际情况请根据项目决定 producer.setDeliveryMode(DeliveryMode.NON_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{ for (int i = 1; i <= 100000; i++){ TextMessage message = session.createTextMessage( "今日天气asda" + i); //发送消息到目的地 System.out.println( "发送:" +message.getText()); producer.send(message); } } } |
接收端程序
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | import org.apache.activemq.ActiveMQConnection;import org.apache.activemq. ActiveMQConnectionFactory;import javax.jms.*;public class revice { public static void main(String[] args) { ConnectionFactory connectionFactory; Connection connection = null ; Session session; Destination destination; //接收者-消费者 MessageConsumer messageConsumer; connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616" ); try { connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession( false ,Session.AUTO_ACKNOWLEDGE); destination = session.createQueue( "test" ); messageConsumer = session.createConsumer(destination); while ( true ){ TextMessage message = (TextMessage) messageConsumer.receive(1000); if ( null != message){ System.out.println( "收到:" +message.getText()); } else { break ; } message.acknowledge(); } } catch (Exception ex){ ex.printStackTrace(); }finally{ try { if ( null != connection){ connection.close(); } } catch (Throwable ig){ } } } } |