ActiveMQ教程（消息发送和接受）

2023-08-10 22:43:46

一环境的搭建

　　version为你的版本号

　　如果你是普通的项目的话，创建一个lib文件夹，导入相应的jar包到你的lib中，jar包为：activemq-all-{version}.jar、log4j-{version}.jar、slf4j-log4j12-{version}.jar,并且bulidpath，将jar加载到你的项目中就可以使用了。

　　如果是web项目，需要将以上的jar包导入到你的web-inf的lib中，将自动加载到你的工程中。

　　如果你是一个mave工程，需要修改你的pom.xml文件，添加相关的依赖：（如下）

　　<dependency>
   　　<groupId>org.apache.activemq</groupId>
   　　<artifactId>activemq-all</artifactId>
   　　<version>{version}</version>
　　</dependency>

　　<dependency>
   　　<groupId>log4j</groupId>
   　　<artifactId>log4j</artifactId>
   　　<version>{version}</version>
　　</dependency>

　　<dependency>
   　　<groupId>org.slf4j</groupId>
   　　<artifactId>slf4j-log4j12</artifactId>
   　　<version>{version}</version>
　　</dependency>

　　相应的版本可以去maven的资源库中下载：http://www.mvnrepository.com。

　　导入相应的jar包之后就可以进行相应的开发了，现在我们开发发送端。创建创建一个普通的java类，名称为ActiveMqUtil

import java.util.Date;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;

public class ActiveMqUtil {

   public static void senderMessage(String message){
       //链接工厂
       ConnectionFactory connectionFactory;
       //创建链接
       Connection connection = null;
       //创建一个session
       Session session;
       //创建目的地
       Destination destination;
       //消息提供者
       MessageProducer messageProducer;
       //构造ConnectionFactory
       connectionFactory = new org.apache.activemq.ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,
               "tcp://localhost:61616");

　　 // 设置用户名和密码，这个用户名和密码在conf目录下的credentials.properties文件中，也可以在activemq.xml中配置，我这里是默认的，所以就注释掉了
       //connectionFactory.setUserName("应户名");
       //connectionFactory.setPassword("密码");
       try {
           //得到连接对象
           connection = connectionFactory.createConnection();
           //启动链接
           connection.start();

　　　　// 创建Session，参数解释：
            // 第一个参数是否使用事务:当消息发送者向消息提供者（即消息代理）发送消息时，消息发送者等待消息代理的确认，没有回应则抛出异常，消息发送程序负责处理这个错误。
            // 第二个参数消息的确认模式：
            // AUTO_ACKNOWLEDGE ：指定消息提供者在每次收到消息时自动发送确认。消息只向目标发送一次，但传输过程中可能因为错误而丢失消息。
            // CLIENT_ACKNOWLEDGE ：由消息接收者确认收到消息，通过调用消息的acknowledge()方法（会通知消息提供者收到了消息）
            // DUPS_OK_ACKNOWLEDGE ：指定消息提供者在消息接收者没有确认发送时重新发送消息（这种确认模式不在乎接收者收到重复的消息）。
           session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
           //获取服务商的消息
           destination = session.createQueue("你的active里面创建的que的名称");
           //得到消息的发送者
            messageProducer = session.createProducer(destination);
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            sendMessage(session, messageProducer,message);
            session.commit();
       } catch (JMSException e) {
           // TODO Auto-generated catch block
           e.printStackTrace();
       }finally{
               try {
                   if(null != connection){
                       connection.close();
                   }
               } catch (JMSException e) {
                   // TODO Auto-generated catch block
                   e.printStackTrace();
           }
       }

   }

   public static void sendMessage(Session session,MessageProducer messageProducer,String messages) throws JMSException{
       TextMessage message = session.createTextMessage(messages);
       //发送消息到服务器
       messageProducer.send(message);
   }
}

之后创建一个测试类，Test

　　public class Test{

　　　　public static void main(String [] args){

　　　　　　ActiveMqUtil.senderMessage("发送者提供的消息!");

　　　　}

运行test的main方法，进行相应的发送。现在你可以访问你安装的activemq的页面去查看相应的信息。这个que的Number Of Consumers会多一条记录，代表发送成功。

现在发送者已经发送成功，现在我们开发接受端。创建一个类，名称为Reciver

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.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Reciver {

   public static void main(String[] args) {
       //创建工厂
       ConnectionFactory connectionFactory;
       //创建connection
       Connection connection = null;
       //创建session
       Session session;
       //创建目的地
       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(false, Session.AUTO_ACKNOWLEDGE);
           //获取服务器上的消息
           destination = session.createQueue("yc-security-mq");
           consumer = session.createConsumer(destination);
           while(true){
               TextMessage message = (TextMessage) consumer.receive(100000);
               if(null != message){
                   System.out.println(message.getText());
               }else{
                   break;
               }
           }
       } catch (JMSException e) {
           // TODO Auto-generated catch block
           e.printStackTrace();
       }finally{
           try{
               if(null != null){
                   connection.close();
               }
           }catch(Exception e){
               e.printStackTrace();
           }
       }
   }

}

运行这个测试类，将会显示：发送者提供的消息!，这时代表接受端成功。在activemq安装的页面上，Messages Dequeued 将会增加一条。

码农公寓

相关文章