ActiveMQ

Posted 2022-11-23 Luther-

上篇介绍了ActiveMQ安装以及使用，下面介绍下如果在程式中使用

首先下载ActiveMQ Jar包

然后创建两个Java class 

消息生成：

package com.htc.question.util;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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;
import org.apache.activemq.ActiveMQConnectionFactory;

public class JMSProducer 
	
	 //默认连接用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    //默认连接密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    //默认连接地址
    private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
    //发送的消息数量
    private static final int SENDNUM = 10;
    
    public static void main(String[] args) 
    	//连接工厂
        ConnectionFactory connectionFactory;
      //连接
        Connection connection = null;        
      //会话 接受或者发送消息的线程
        Session session;
      //消息的目的地
        Destination destination;
      //消息生产者
        MessageProducer messageProducer;
      //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
        
        try
        	//通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //创建一个名称为HelloWorld的消息队列
            destination = session.createQueue("HelloWorld");
            //创建消息生产者
            messageProducer = session.createProducer(destination);
            //发送消息
            sendMessage(session, messageProducer);

            session.commit();
        catch (Exception e) 
			// TODO: handle exception
		finally			
			if(connection != null)
                try 
                    connection.close();
                 catch (JMSException e) 
                    e.printStackTrace();
                
            
		
	
    
    
    public static void sendMessage(Session session,MessageProducer messageProducer) throws Exception
        for (int i = 0; i < JMSProducer.SENDNUM; i++) 
            //创建一条文本消息 
            TextMessage message = session.createTextMessage("ActiveMQ 发送消息" +i);
            System.out.println("发送消息：Activemq 发送消息" + i);
            //通过消息生产者发出消息 
            messageProducer.send(message);
        

    
	

在HelloWord消息队列中有10条消息在队列中，Messages Enqueued 是生产消息的总数，Messages Dequeued 是消费消息的总数，Number Of Pending Messages 是队列中待消费的消息数量，由此可以看出刚才执行的java 代码块生成了10条信息。

接下来我们编写消费代码

消费代码：

package com.htc.question.util;

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 JMSConsumer 
	
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
    private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址
    
    public static void main(String[] args) 
    	 ConnectionFactory connectionFactory;//连接工厂
         Connection connection = null;//连接

         Session session;//会话 接受或者发送消息的线程
         Destination destination;//消息的目的地

         MessageConsumer messageConsumer;//消息的消费者
         
       //实例化连接工厂
         connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
         
         try 
             //通过连接工厂获取连接
             connection = connectionFactory.createConnection();
             //启动连接
             connection.start();
             //创建session
             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //创建一个连接HelloWorld的消息队列
             destination = session.createQueue("HelloWorld");
             //创建消息消费者
             messageConsumer = session.createConsumer(destination);
             
             while (true) 
                 TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
                 if(textMessage != null)
                     System.out.println("收到的消息:" + textMessage.getText());
                 else 
                     break;
                 
             

          catch (JMSException e) 
             e.printStackTrace();
         

	

由此可以看出消息被消费。Number Of Consumers 是消费者。

这是实现点对点消息模式，以及发送同步的消息。