ActiveMQ 部署及发送接收消息

时间  2021-08-13 标签 java web apache windows session dom tcp ide 测试 url

ActiveMQ 部署及发送接收消息java

1、           下载

下载地址：http://activemq.apache.org/ 我这里使用的版本为当前最新5.8.0。web

下载版本有Windows和Linux两个版本，且都分为32位和64位。根据本身须要选择下载。apache

2、           安装

我这里下载的为windows的32位版本（apache-activemq-5.8.0-bin.zip），下载后直接解压到须要安装的目录或在直接解压到当前目录也可，解压完安装也完成。windows


解压后目录如上图，里面包含了示例和文档，及全部的jar包。session

 

3、           运行

进入到bin目录（apache-activemq-5.8.0\bin），双击activemq.bat，就会运行，运行截图以下：dom



此时表示ActiveMQ已经在运行了，固然正常生产环境下能够设置做为服务在后台运行，而且随系统启动而启动。tcp

4、           测试

ActiveMQ自带了一套管理系统，访问http://localhost:8161/admin/，会出现须要输入用户名和密码的页面以下：ide



默认用户名和密码都是admin，进入后则为主界面：测试



在这个界面上，咱们能够管理队列及其余的一些功能，为了下面的继续，咱们在这里建立一个Queue和一个Topic。url



点击目录上的Queues进入建立Queue页面，输入Queue名称，点击Create后下面就建立了G2Queue的queue队列。




这里也能够不用这样手工建立，在发送端指定了一个Queue或Topic名字后，会自动建立一个队列，如上面的choice.queue和FirstQueue都是我测试程序时，程序里面指定的Queue名称，自动建立的。

一样的方式建立一个Topic，以下：



5、           发送消息

建立一个新的项目，我这里是建立的webproject名称为ActiveMQ，引入ActiveMQ的jar包，整个工程结构以下：


 

此段代码从网上直接copy，只是稍做修改：

[java] view plaincopy

  1. import java.util.Random;  

  2.    

  3. import javax.jms.Connection;  

  4. import javax.jms.ConnectionFactory;  

  5. import javax.jms.DeliveryMode;  

  6. import javax.jms.Destination;  

  7. import javax.jms.JMSException;  

  8. import javax.jms.MessageProducer;  

  9. import javax.jms.Session;  

  10. import javax.jms.TextMessage;  

  11.    

  12. import org.apache.activemq.ActiveMQConnectionFactory;  

  13.    

  14. public class SendMessage {  

  15.     private static final String url = "tcp://localhost:61616";  

  16.     private static final String QUEUE_NAME = "G2Queue";  

  17.    

  18.     public void sendMessage() throwsJMSException {  

  19.        // JMS 客户端到JMSProvider 的链接  

  20.        Connection connection = null;  

  21.        try {  

  22.            // 链接工厂，JMS 用它建立链接  

  23.            // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar  

  24.            ConnectionFactory connectionFactory = newActiveMQConnectionFactory(url);  

  25.            connection = (Connection)connectionFactory.createConnection();  

  26.            // 启动链接  

  27.            connection.start();  

  28.            //Session：发送或接收消息的线程  

  29.            // 获取session  

  30.            Session session = (Session) connection.createSession(false,  

  31.                   Session.AUTO_ACKNOWLEDGE);  

  32.            // 消息的目的地，消息发送到那个队列  

  33.            Destination destination = session.createQueue(QUEUE_NAME);  

  34.            //MessageProducer：消息发送者（生产者）  

  35.            // 建立消息发送者  

  36.            MessageProducer producer =session.createProducer(destination);  

  37.            // 设置是否持久化  

  38.            //DeliveryMode.NON_PERSISTENT：不持久化  

  39.            //DeliveryMode.PERSISTENT：持久化  

  40.            producer.setDeliveryMode(DeliveryMode.PERSISTENT);  

  41.             

  42.            String msg = "";  

  43.            int i = 0;  

  44.         do {  

  45.             msg = "第"+i + "次发送的消息："+new Random();  

  46.                 TextMessagemessage = session.createTextMessage(msg);  

  47.                 Thread.sleep(1000);  

  48.                 // 发送消息到目的地方  

  49.                producer.send(message);  

  50.                 System.out.println("发送消息：" +msg);  

  51.                 i++;  

  52.         } while (i<1000);  

  53.        } catch (Exception e) {  

  54.            e.printStackTrace();  

  55.        }  

  56.     }  

  57.    

  58.     public static void main(String[] args) {  

  59.        SendMessage sndMsg = newSendMessage();  

  60.        try {  

  61.            sndMsg.sendMessage();  

  62.        } catch (Exception ex) {  

  63.            System.out.println(ex.toString());  

  64.        }  

  65.     }  

  66. }  




运行结果以下：

6、           接收消息


[java] view plaincopy

  1. package cn.g2room.mq.test;  

  2.    

  3. import javax.jms.Connection;  

  4. import javax.jms.ConnectionFactory;  

  5. import javax.jms.Destination;  

  6. import javax.jms.JMSException;  

  7. import javax.jms.Message;  

  8. import javax.jms.MessageConsumer;  

  9. import javax.jms.Session;  

  10. import javax.jms.TextMessage;  

  11.    

  12. import org.apache.activemq.ActiveMQConnectionFactory;  

  13.    

  14. /** 

  15.  * 消息接收类 

  16.  * 

  17.  * @createTime:Apr 7, 2013 5:11:11 PM 

  18.  * @author:<a href="mailto:252909344@qq.com">迷蝶</a> 

  19.  * @version:0.1 

  20.  *@lastVersion: 0.1 

  21.  * @updateTime: 

  22.  *@updateAuthor: <a href="mailto:252909344@qq.com">迷蝶</a> 

  23.  * @changesSum: 

  24.  * 

  25.  */  

  26. public class ReceiveMessage {  

  27.          privatestatic final String url = "tcp://localhost:61616";  

  28.          privatestatic final String QUEUE_NAME = "G2Queue";  

  29.    

  30.          publicvoid receiveMessage() {  

  31.                    Connectionconnection = null;  

  32.                    try{  

  33.                             try{  

  34.                                      ConnectionFactoryconnectionFactory = new ActiveMQConnectionFactory(  

  35.                                                         url);  

  36.                                      connection= connectionFactory.createConnection();  

  37.                             }catch (Exception e) {  

  38.                                      System.out.println(e.toString());  

  39.                             }  

  40.                             connection.start();  

  41.                             Sessionsession = connection.createSession(false,  

  42.                                                Session.AUTO_ACKNOWLEDGE);  

  43.                             Destinationdestination = session.createQueue(QUEUE_NAME);  

  44.                             //消息接收者，也就是消费者  

  45.                             MessageConsumerconsumer = session.createConsumer(destination);  

  46.                              

  47.                             consumeMessagesAndClose(connection,session, consumer);  

  48.                    }catch (Exception e) {  

  49.                             System.out.println(e.toString());  

  50.                    }  

  51.          }  

  52.          /** 

  53.           * 接收和关闭消息，如遇到消息内容为close则，关闭链接 

  54.           * 

  55.           * @param connection   JMS 客户端到JMSProvider 的链接 

  56.           * @param session                   发送或接收消息的线程 

  57.           * @param consumer              消息接收对象 

  58.           * @throws JMSException 

  59.           * @auther <ahref="mailto:252909344@qq.com">迷蝶</a> 

  60.           * Apr 8, 2013 10:31:55 AM 

  61.           */  

  62.          protectedvoid consumeMessagesAndClose(Connection connection,  

  63.                             Sessionsession, MessageConsumer consumer) throws JMSException {  

  64.                    do{  

  65.                             Messagemessage = consumer.receive(1000);  

  66.                             if("close".equals(message)){  

  67.                                      consumer.close();  

  68.                                      session.close();  

  69.                                      connection.close();  

  70.                             }  

  71.                             if(message != null) {  

  72.                                      onMessage(message);  

  73.                             }  

  74.                    }while (true);  

  75.                     

  76.          }  

  77.    

  78.          publicvoid onMessage(Message message) {  

  79.                    try{  

  80.                             if(message instanceof TextMessage) {  

  81.                                      TextMessagetxtMsg = (TextMessage) message;  

  82.                                      Stringmsg = txtMsg.getText();  

  83.                                      System.out.println("Received:" + msg);  

  84.                             }  

  85.                    }catch (Exception e) {  

  86.                             e.printStackTrace();  

  87.                    }  

  88.    

  89.          }  

  90.    

  91.          publicstatic void main(String args[]) {  

  92.                    ReceiveMessagerm = new ReceiveMessage();  

  93.                    rm.receiveMessage();  

  94.          }  

  95. }  




运行结果以下：


 

联系我们 沪ICP备13005482号-10